You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/03/09 16:49:10 UTC
(iceberg) branch main updated: Flink: Backport Add the possibility to use Coordinator Lock when using Flink SQL to 2.0 and 1.20 (#15562)
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new afb58aa341 Flink: Backport Add the possibility to use Coordinator Lock when using Flink SQL to 2.0 and 1.20 (#15562)
afb58aa341 is described below
commit afb58aa341d055a6bef81a54d1e9e20d07b7e075
Author: GuoYu <51...@qq.com>
AuthorDate: Tue Mar 10 00:48:55 2026 +0800
Flink: Backport Add the possibility to use Coordinator Lock when using Flink SQL to 2.0 and 1.20 (#15562)
Backports #15459
---
.../maintenance/operator/LockRemoverOperator.java | 7 +++-
.../org/apache/iceberg/flink/sink/IcebergSink.java | 16 +++++---
.../flink/TestFlinkTableSinkCompaction.java | 46 ++++++++++++++++++----
.../operator/TestLockRemoverOperation.java | 21 ++++++++++
.../iceberg/flink/sink/TestIcebergSinkCompact.java | 36 +++++++++++------
.../maintenance/operator/LockRemoverOperator.java | 7 +++-
.../org/apache/iceberg/flink/sink/IcebergSink.java | 16 +++++---
.../flink/TestFlinkTableSinkCompaction.java | 46 ++++++++++++++++++----
.../operator/TestLockRemoverOperation.java | 21 ++++++++++
.../iceberg/flink/sink/TestIcebergSinkCompact.java | 36 +++++++++++------
10 files changed, 200 insertions(+), 52 deletions(-)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java
index c4da397e4c..bee95f5003 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java
@@ -104,8 +104,11 @@ class LockRemoverOperator extends AbstractStreamOperator<Void>
@Override
public void processWatermark(Watermark mark) throws Exception {
- operatorEventGateway.sendEventToCoordinator(
- new LockReleaseEvent(tableName, mark.getTimestamp()));
+ if (Watermark.MAX_WATERMARK.getTimestamp() != mark.getTimestamp()) {
+ operatorEventGateway.sendEventToCoordinator(
+ new LockReleaseEvent(tableName, mark.getTimestamp()));
+ }
+
super.processWatermark(mark);
}
}
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
index 2787a3aaa9..dd20853f05 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
@@ -29,6 +29,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.function.Function;
+import org.apache.commons.lang3.StringUtils;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.api.common.SupportsConcurrentExecutionAttempts;
import org.apache.flink.api.common.functions.FlatMapFunction;
@@ -76,7 +77,6 @@ import org.apache.iceberg.flink.maintenance.api.LockConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFiles;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
import org.apache.iceberg.flink.maintenance.api.TableMaintenance;
-import org.apache.iceberg.flink.maintenance.api.TriggerLockFactory;
import org.apache.iceberg.flink.maintenance.operator.LockFactoryBuilder;
import org.apache.iceberg.flink.maintenance.operator.TableChange;
import org.apache.iceberg.flink.sink.shuffle.DataStatisticsOperatorFactory;
@@ -270,12 +270,18 @@ public class IcebergSink
RewriteDataFiles.builder().config(rewriteDataFilesConfig);
LockConfig lockConfig = flinkMaintenanceConfig.createLockConfig();
- TriggerLockFactory triggerLockFactory = LockFactoryBuilder.build(lockConfig, table.name());
String tableMaintenanceUid = String.format("TableMaintenance : %s", suffix);
TableMaintenance.Builder builder =
- TableMaintenance.forChangeStream(tableChangeStream, tableLoader, triggerLockFactory)
- .uidSuffix(tableMaintenanceUid)
- .add(rewriteBuilder);
+ StringUtils.isNotEmpty(lockConfig.lockType())
+ ? TableMaintenance.forChangeStream(
+ tableChangeStream,
+ tableLoader,
+ LockFactoryBuilder.build(lockConfig, table.name()))
+ .uidSuffix(tableMaintenanceUid)
+ .add(rewriteBuilder)
+ : TableMaintenance.forChangeStream(tableChangeStream, tableLoader)
+ .uidSuffix(tableMaintenanceUid)
+ .add(rewriteBuilder);
builder
.rateLimit(Duration.ofSeconds(flinkMaintenanceConfig.rateLimit()))
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
index 03d96ac2c5..b0c4f595aa 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
@@ -39,11 +39,13 @@ import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.maintenance.api.LockConfig;
import org.apache.iceberg.flink.source.BoundedTableFactory;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
@@ -51,7 +53,9 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestFlinkTableSinkCompaction extends CatalogTestBase {
private static final TypeInformation<Row> ROW_TYPE_INFO =
@@ -75,15 +79,31 @@ public class TestFlinkTableSinkCompaction extends CatalogTestBase {
+ "'flink-maintenance.rewrite.rewrite-all'='true',"
+ "'flink-maintenance.rewrite.schedule.data-file-size'='1',"
+ "'flink-maintenance.lock-check-delay-seconds'='60'";
+ private static final String TABLE_PROPERTIES_COORDINATOR =
+ "'flink-maintenance.rewrite.rewrite-all'='true',"
+ + "'flink-maintenance.rewrite.schedule.data-file-size'='1',"
+ + "'flink-maintenance.lock-check-delay-seconds'='60'";
@Parameter(index = 2)
private boolean userSqlHint;
- @Parameters(name = "catalogName={0}, baseNamespace={1}, userSqlHint={2}")
+ @Parameter(index = 3)
+ private String lockType;
+
+ @Parameters(name = "catalogName={0}, baseNamespace={1}, userSqlHint={2}, lockType={3}")
public static List<Object[]> parameters() {
return Arrays.asList(
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), true},
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), false});
+ new Object[] {
+ "testhadoop_basenamespace", Namespace.of("l0", "l1"), true, LockConfig.JdbcLockConfig.JDBC
+ },
+ new Object[] {
+ "testhadoop_basenamespace",
+ Namespace.of("l0", "l1"),
+ false,
+ LockConfig.JdbcLockConfig.JDBC
+ },
+ new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), true, ""},
+ new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), false, ""});
}
@Override
@@ -118,7 +138,13 @@ public class TestFlinkTableSinkCompaction extends CatalogTestBase {
if (userSqlHint) {
sql("CREATE TABLE %s (id int, data varchar)", TABLE_NAME);
} else {
- sql("CREATE TABLE %s (id int, data varchar) with (%s)", TABLE_NAME, TABLE_PROPERTIES);
+ if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
+ sql("CREATE TABLE %s (id int, data varchar) with (%s)", TABLE_NAME, TABLE_PROPERTIES);
+ } else {
+ sql(
+ "CREATE TABLE %s (id int, data varchar) with (%s)",
+ TABLE_NAME, TABLE_PROPERTIES_COORDINATOR);
+ }
}
icebergTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
@@ -144,9 +170,15 @@ public class TestFlinkTableSinkCompaction extends CatalogTestBase {
// Redirect the records from source table to destination table.
if (userSqlHint) {
- sql(
- "INSERT INTO %s /*+ OPTIONS(%s) */ SELECT id,data from sourceTable",
- TABLE_NAME, TABLE_PROPERTIES);
+ if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
+ sql(
+ "INSERT INTO %s /*+ OPTIONS(%s) */ SELECT id,data from sourceTable",
+ TABLE_NAME, TABLE_PROPERTIES);
+ } else {
+ sql(
+ "INSERT INTO %s /*+ OPTIONS(%s) */ SELECT id,data from sourceTable",
+ TABLE_NAME, TABLE_PROPERTIES_COORDINATOR);
+ }
} else {
sql("INSERT INTO %s SELECT id,data from sourceTable", TABLE_NAME);
}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java
index 9ba95cab30..a0b615e463 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.iceberg.flink.maintenance.api.TaskResult;
@@ -89,6 +90,26 @@ class TestLockRemoverOperation extends OperatorTestBase {
}
}
+ @Test
+ void testProcessMaxWaterMark() throws Exception {
+ MockOperatorEventGateway mockGateway = new MockOperatorEventGateway();
+ LockRemoverOperator operator =
+ new LockRemoverOperator(null, mockGateway, DUMMY_TASK_NAME, Lists.newArrayList(TASKS[0]));
+ try (OneInputStreamOperatorTestHarness<TaskResult, Void> testHarness =
+ createHarness(operator)) {
+
+ testHarness.processElement(
+ new StreamRecord<>(new TaskResult(0, 0L, true, Lists.newArrayList())));
+ assertThat(mockGateway.getEventsSent()).hasSize(0);
+
+ testHarness.processWatermark(WATERMARK);
+ assertThat(mockGateway.getEventsSent()).hasSize(1);
+
+ testHarness.processWatermark(Watermark.MAX_WATERMARK);
+ assertThat(mockGateway.getEventsSent()).hasSize(1);
+ }
+ }
+
@Test
void testMetrics() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java
index b84d21d020..48d5f9382a 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java
@@ -48,9 +48,11 @@ import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.FieldSource;
class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
+ private static final String[] LOCK_TYPES = new String[] {LockConfig.JdbcLockConfig.JDBC, ""};
private Map<String, String> flinkConf;
@@ -58,14 +60,7 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
void before() throws IOException {
this.flinkConf = Maps.newHashMap();
flinkConf.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
- flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
- flinkConf.put(
- LockConfig.JdbcLockConfig.JDBC_URI_OPTION.key(),
- "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
- flinkConf.put(LockConfig.LOCK_ID_OPTION.key(), "test-lock-id");
flinkConf.put(RewriteDataFilesConfig.SCHEDULE_ON_DATA_FILE_SIZE, "1");
-
- flinkConf.put(LockConfig.JdbcLockConfig.JDBC_INIT_LOCK_TABLE_OPTION.key(), "true");
flinkConf.put(RewriteDataFilesConfig.PREFIX + SizeBasedFileRewritePlanner.REWRITE_ALL, "true");
table =
@@ -85,8 +80,10 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
tableLoader = CATALOG_EXTENSION.tableLoader();
}
- @Test
- public void testCompactFileE2e() throws Exception {
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testCompactFileE2e(String lockType) throws Exception {
+ setupLockConfig(lockType);
List<Row> rows = Lists.newArrayList(Row.of(1, "hello"), Row.of(2, "world"), Row.of(3, "foo"));
DataStream<RowData> dataStream =
env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
@@ -122,8 +119,10 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
return dataFiles;
}
- @Test
- public void testTableMaintenanceOperatorAdded() {
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testTableMaintenanceOperatorAdded(String lockType) {
+ setupLockConfig(lockType);
List<Row> rows = Lists.newArrayList(Row.of(1, "hello"), Row.of(2, "world"), Row.of(3, "foo"));
DataStream<RowData> dataStream =
env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
@@ -146,4 +145,17 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
assertThat(containRewrite).isTrue();
}
+
+ private void setupLockConfig(String lockType) {
+ if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
+ flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
+ flinkConf.put(
+ LockConfig.JdbcLockConfig.JDBC_URI_OPTION.key(),
+ "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
+ flinkConf.put(LockConfig.LOCK_ID_OPTION.key(), "test-lock-id");
+ flinkConf.put(LockConfig.JdbcLockConfig.JDBC_INIT_LOCK_TABLE_OPTION.key(), "true");
+ } else {
+ flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), "");
+ }
+ }
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java
index 51aae657e6..d5c1be2e32 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java
@@ -105,8 +105,11 @@ class LockRemoverOperator extends AbstractStreamOperator<Void>
@Override
public void processWatermark(Watermark mark) throws Exception {
- operatorEventGateway.sendEventToCoordinator(
- new LockReleaseEvent(tableName, mark.getTimestamp()));
+ if (Watermark.MAX_WATERMARK.getTimestamp() != mark.getTimestamp()) {
+ operatorEventGateway.sendEventToCoordinator(
+ new LockReleaseEvent(tableName, mark.getTimestamp()));
+ }
+
super.processWatermark(mark);
}
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
index 65b6f62527..96c947a853 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
@@ -29,6 +29,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.function.Function;
+import org.apache.commons.lang3.StringUtils;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.api.common.SupportsConcurrentExecutionAttempts;
import org.apache.flink.api.common.functions.FlatMapFunction;
@@ -77,7 +78,6 @@ import org.apache.iceberg.flink.maintenance.api.LockConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFiles;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
import org.apache.iceberg.flink.maintenance.api.TableMaintenance;
-import org.apache.iceberg.flink.maintenance.api.TriggerLockFactory;
import org.apache.iceberg.flink.maintenance.operator.LockFactoryBuilder;
import org.apache.iceberg.flink.maintenance.operator.TableChange;
import org.apache.iceberg.flink.sink.shuffle.DataStatisticsOperatorFactory;
@@ -270,12 +270,18 @@ public class IcebergSink
RewriteDataFiles.builder().config(rewriteDataFilesConfig);
LockConfig lockConfig = flinkMaintenanceConfig.createLockConfig();
- TriggerLockFactory triggerLockFactory = LockFactoryBuilder.build(lockConfig, table.name());
String tableMaintenanceUid = String.format("TableMaintenance : %s", suffix);
TableMaintenance.Builder builder =
- TableMaintenance.forChangeStream(tableChangeStream, tableLoader, triggerLockFactory)
- .uidSuffix(tableMaintenanceUid)
- .add(rewriteBuilder);
+ StringUtils.isNotEmpty(lockConfig.lockType())
+ ? TableMaintenance.forChangeStream(
+ tableChangeStream,
+ tableLoader,
+ LockFactoryBuilder.build(lockConfig, table.name()))
+ .uidSuffix(tableMaintenanceUid)
+ .add(rewriteBuilder)
+ : TableMaintenance.forChangeStream(tableChangeStream, tableLoader)
+ .uidSuffix(tableMaintenanceUid)
+ .add(rewriteBuilder);
builder
.rateLimit(Duration.ofSeconds(flinkMaintenanceConfig.rateLimit()))
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
index 03d96ac2c5..b0c4f595aa 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java
@@ -39,11 +39,13 @@ import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.maintenance.api.LockConfig;
import org.apache.iceberg.flink.source.BoundedTableFactory;
import org.apache.iceberg.flink.source.BoundedTestSource;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
@@ -51,7 +53,9 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+@ExtendWith(ParameterizedTestExtension.class)
public class TestFlinkTableSinkCompaction extends CatalogTestBase {
private static final TypeInformation<Row> ROW_TYPE_INFO =
@@ -75,15 +79,31 @@ public class TestFlinkTableSinkCompaction extends CatalogTestBase {
+ "'flink-maintenance.rewrite.rewrite-all'='true',"
+ "'flink-maintenance.rewrite.schedule.data-file-size'='1',"
+ "'flink-maintenance.lock-check-delay-seconds'='60'";
+ private static final String TABLE_PROPERTIES_COORDINATOR =
+ "'flink-maintenance.rewrite.rewrite-all'='true',"
+ + "'flink-maintenance.rewrite.schedule.data-file-size'='1',"
+ + "'flink-maintenance.lock-check-delay-seconds'='60'";
@Parameter(index = 2)
private boolean userSqlHint;
- @Parameters(name = "catalogName={0}, baseNamespace={1}, userSqlHint={2}")
+ @Parameter(index = 3)
+ private String lockType;
+
+ @Parameters(name = "catalogName={0}, baseNamespace={1}, userSqlHint={2}, lockType={3}")
public static List<Object[]> parameters() {
return Arrays.asList(
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), true},
- new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), false});
+ new Object[] {
+ "testhadoop_basenamespace", Namespace.of("l0", "l1"), true, LockConfig.JdbcLockConfig.JDBC
+ },
+ new Object[] {
+ "testhadoop_basenamespace",
+ Namespace.of("l0", "l1"),
+ false,
+ LockConfig.JdbcLockConfig.JDBC
+ },
+ new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), true, ""},
+ new Object[] {"testhadoop_basenamespace", Namespace.of("l0", "l1"), false, ""});
}
@Override
@@ -118,7 +138,13 @@ public class TestFlinkTableSinkCompaction extends CatalogTestBase {
if (userSqlHint) {
sql("CREATE TABLE %s (id int, data varchar)", TABLE_NAME);
} else {
- sql("CREATE TABLE %s (id int, data varchar) with (%s)", TABLE_NAME, TABLE_PROPERTIES);
+ if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
+ sql("CREATE TABLE %s (id int, data varchar) with (%s)", TABLE_NAME, TABLE_PROPERTIES);
+ } else {
+ sql(
+ "CREATE TABLE %s (id int, data varchar) with (%s)",
+ TABLE_NAME, TABLE_PROPERTIES_COORDINATOR);
+ }
}
icebergTable = validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, TABLE_NAME));
@@ -144,9 +170,15 @@ public class TestFlinkTableSinkCompaction extends CatalogTestBase {
// Redirect the records from source table to destination table.
if (userSqlHint) {
- sql(
- "INSERT INTO %s /*+ OPTIONS(%s) */ SELECT id,data from sourceTable",
- TABLE_NAME, TABLE_PROPERTIES);
+ if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
+ sql(
+ "INSERT INTO %s /*+ OPTIONS(%s) */ SELECT id,data from sourceTable",
+ TABLE_NAME, TABLE_PROPERTIES);
+ } else {
+ sql(
+ "INSERT INTO %s /*+ OPTIONS(%s) */ SELECT id,data from sourceTable",
+ TABLE_NAME, TABLE_PROPERTIES_COORDINATOR);
+ }
} else {
sql("INSERT INTO %s SELECT id,data from sourceTable", TABLE_NAME);
}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java
index 9ba95cab30..a0b615e463 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestLockRemoverOperation.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.iceberg.flink.maintenance.api.TaskResult;
@@ -89,6 +90,26 @@ class TestLockRemoverOperation extends OperatorTestBase {
}
}
+ @Test
+ void testProcessMaxWaterMark() throws Exception {
+ MockOperatorEventGateway mockGateway = new MockOperatorEventGateway();
+ LockRemoverOperator operator =
+ new LockRemoverOperator(null, mockGateway, DUMMY_TASK_NAME, Lists.newArrayList(TASKS[0]));
+ try (OneInputStreamOperatorTestHarness<TaskResult, Void> testHarness =
+ createHarness(operator)) {
+
+ testHarness.processElement(
+ new StreamRecord<>(new TaskResult(0, 0L, true, Lists.newArrayList())));
+ assertThat(mockGateway.getEventsSent()).hasSize(0);
+
+ testHarness.processWatermark(WATERMARK);
+ assertThat(mockGateway.getEventsSent()).hasSize(1);
+
+ testHarness.processWatermark(Watermark.MAX_WATERMARK);
+ assertThat(mockGateway.getEventsSent()).hasSize(1);
+ }
+ }
+
@Test
void testMetrics() throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java
index b84d21d020..48d5f9382a 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java
@@ -48,9 +48,11 @@ import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.FieldSource;
class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
+ private static final String[] LOCK_TYPES = new String[] {LockConfig.JdbcLockConfig.JDBC, ""};
private Map<String, String> flinkConf;
@@ -58,14 +60,7 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
void before() throws IOException {
this.flinkConf = Maps.newHashMap();
flinkConf.put(FlinkWriteOptions.COMPACTION_ENABLE.key(), "true");
- flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
- flinkConf.put(
- LockConfig.JdbcLockConfig.JDBC_URI_OPTION.key(),
- "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
- flinkConf.put(LockConfig.LOCK_ID_OPTION.key(), "test-lock-id");
flinkConf.put(RewriteDataFilesConfig.SCHEDULE_ON_DATA_FILE_SIZE, "1");
-
- flinkConf.put(LockConfig.JdbcLockConfig.JDBC_INIT_LOCK_TABLE_OPTION.key(), "true");
flinkConf.put(RewriteDataFilesConfig.PREFIX + SizeBasedFileRewritePlanner.REWRITE_ALL, "true");
table =
@@ -85,8 +80,10 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
tableLoader = CATALOG_EXTENSION.tableLoader();
}
- @Test
- public void testCompactFileE2e() throws Exception {
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testCompactFileE2e(String lockType) throws Exception {
+ setupLockConfig(lockType);
List<Row> rows = Lists.newArrayList(Row.of(1, "hello"), Row.of(2, "world"), Row.of(3, "foo"));
DataStream<RowData> dataStream =
env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
@@ -122,8 +119,10 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
return dataFiles;
}
- @Test
- public void testTableMaintenanceOperatorAdded() {
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testTableMaintenanceOperatorAdded(String lockType) {
+ setupLockConfig(lockType);
List<Row> rows = Lists.newArrayList(Row.of(1, "hello"), Row.of(2, "world"), Row.of(3, "foo"));
DataStream<RowData> dataStream =
env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
@@ -146,4 +145,17 @@ class TestIcebergSinkCompact extends TestFlinkIcebergSinkBase {
assertThat(containRewrite).isTrue();
}
+
+ private void setupLockConfig(String lockType) {
+ if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
+ flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
+ flinkConf.put(
+ LockConfig.JdbcLockConfig.JDBC_URI_OPTION.key(),
+ "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
+ flinkConf.put(LockConfig.LOCK_ID_OPTION.key(), "test-lock-id");
+ flinkConf.put(LockConfig.JdbcLockConfig.JDBC_INIT_LOCK_TABLE_OPTION.key(), "true");
+ } else {
+ flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), "");
+ }
+ }
}