Flink: Compact in sink v2 support Coordinator Lock by Guosmilesmile · Pull Request #15459 · apache/iceberg

Guosmilesmile · 2026-02-27T09:04:23Z

We already support the Coordinator Lock, but it hasn’t been introduced into Sink Table Maintenance yet. This PR adds support for configuring the Coordinator Lock in Sink Table Maintenance.

Guosmilesmile · 2026-02-27T09:05:13Z

...link/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManagerOperator.java

-    Preconditions.checkArgument(lockTime != null, "Lock time is null, Can't release lock");
+    if (lockTime == null) {
+      LOG.warn("Lock time is null, Can't release lock");
+      return;
+    }


When we meet max watermark case, this will error, so open a new pr to deal with it .
#15458

I've left a comment on that PR.

Thanks for pointing it out, I have change the way

mxm

Thanks @Guosmilesmile!

docs/docs/flink-maintenance.md

mxm · 2026-03-04T14:09:13Z

docs/docs/flink-maintenance.md

 env.execute("Table Maintenance Job");
 ```

+Use Coordinator Lock


Suggested change

Use Coordinator Lock

#### Managing table locking via Flink

mxm · 2026-03-04T14:10:53Z

docs/docs/flink-maintenance.md

+StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+TableLoader tableLoader = TableLoader.fromCatalog(
+    CatalogLoader.hive("my_catalog", configuration, properties),  
+    TableIdentifier.of("database", "table")
+);
+
+TableMaintenance.forTable(env, tableLoader)
+    .uidSuffix("my-maintenance-job")
+    .rateLimit(Duration.ofMinutes(10))
+    .lockCheckDelay(Duration.ofSeconds(10))
+    .add(ExpireSnapshots.builder()
+        .scheduleOnCommitCount(10)
+        .maxSnapshotAge(Duration.ofMinutes(10))
+        .retainLast(5)
+        .deleteBatchSize(5)
+        .parallelism(8))
+    .add(RewriteDataFiles.builder()
+        .scheduleOnDataFileCount(10)
+        .targetFileSizeBytes(128 * 1024 * 1024)
+        .partialProgressEnabled(true)
+        .partialProgressMaxCommits(10))
+    .append();
+
+env.execute("Table Maintenance Job");


Everything except line 184 is identical (no lock parameter). Maybe consolidate the two sections and just explain that the builder can either be

TableMaintenance.forTable(env, tableLoader, lockFactory)

or

TableMaintenance.forTable(env, tableLoader)

Right, I combine two part code

mxm · 2026-03-04T14:11:08Z

...link/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManagerOperator.java

-    Preconditions.checkArgument(lockTime != null, "Lock time is null, Can't release lock");
+    if (lockTime == null) {
+      LOG.warn("Lock time is null, Can't release lock");
+      return;
+    }


I've left a comment on that PR.

mxm · 2026-03-04T14:11:59Z

docs/docs/flink-maintenance.md

+
+| Key | Description          | Default |
+|-----|----------------------|---------|
+| `flink-maintenance.lock.type` | Set to `` or not set |  |


Would it make sense to have this default to flink? (which will be using the coordinator)

If the coordinator lock will become the default later and the other lock options are removed, I’d prefer not to require users to set this parameter here, so it’s easier and more consistent with their future usage habits.

mxm · 2026-03-05T14:46:18Z

...1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemoverOperator.java

+    if (!Watermark.MAX_WATERMARK.equals(mark)) {
+      operatorEventGateway.sendEventToCoordinator(
+          new LockReleaseEvent(tableName, mark.getTimestamp()));
+    }


Let's compare the timestamp, not the object.

Ok，make it compare with timestamp.

pvary · 2026-03-06T10:08:33Z

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkCompact.java

  }

-  @Test
+  @TestTemplate


In one of my PRs it was suggested not to use TestTemplate as it was only introduced for backward compatibility reasons.

I was pointed to use @ParameterizedTest for every test instead:

@ParameterizedTest @FieldSource("FILE_FORMATS")

pvary · 2026-03-06T10:12:14Z

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java

          + "'flink-maintenance.rewrite.schedule.data-file-size'='1',"
          + "'flink-maintenance.lock-check-delay-seconds'='60'";
+  private static final String TABLE_PROPERTIES_COORDINATOR =
+      "'flink-maintenance.lock.jdbc.init-lock-table'='true',"


Why do we need this?

pvary · 2026-03-06T10:12:54Z

docs/docs/flink-maintenance.md


+#### Flink-maintained lock
+
+Maintain the lock within Flink itself. This does not require configuring external systems. One prerequisite is that there are no parallel table maintenance jobs for a given table.


Should this be?

The only prerequisite is that there are no parallel table maintenance jobs for a given table.

pvary · 2026-03-06T10:13:22Z

docs/docs/flink-maintenance.md

    jdbcProps                                   // JDBC connection properties
 );

+// Option 1: With external lock factory


Shall we mention that we plan to deprecate Option 1?

Do we plan to do it?

pvary · 2026-03-06T10:16:11Z

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java

    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, "jdbc"},


Could this be LockConfig.JdbcLockConfig.JDBC?

pvary · 2026-03-06T10:16:25Z

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkCompaction.java

-      sql(
-          "INSERT INTO %s /*+ OPTIONS(%s) */ SELECT id,data from sourceTable",
-          TABLE_NAME, TABLE_PROPERTIES);
+      if (lockType.equals("jdbc")) {


Could this be: LockConfig.JdbcLockConfig.JDBC?

github-actions bot added flink docs labels Feb 27, 2026

Guosmilesmile commented Feb 27, 2026

View reviewed changes

Flink: Compact in sink v2 support Coordinator Lock

b1262dd

Guosmilesmile force-pushed the table-maintance-sink-coordinator branch from 9e40643 to b1262dd Compare February 27, 2026 09:08

mxm reviewed Mar 4, 2026

View reviewed changes

LockRemoverOperator handle MAX_WATERMARK

3a7233b

Guosmilesmile force-pushed the table-maintance-sink-coordinator branch 2 times, most recently from e15c09d to 3ae3a07 Compare March 5, 2026 05:26

update doc

7193f0b

Guosmilesmile force-pushed the table-maintance-sink-coordinator branch from 3ae3a07 to 7193f0b Compare March 5, 2026 05:26

mxm mentioned this pull request Mar 5, 2026

Flink: Coordinator Lock release avoid send max watermark when shutting down #15458

Open

mxm approved these changes Mar 5, 2026

View reviewed changes

change compare to long

c78e07e

pvary reviewed Mar 6, 2026

View reviewed changes


		#### Flink-maintained lock

		Maintain the lock within Flink itself. This does not require configuring external systems. One prerequisite is that there are no parallel table maintenance jobs for a given table.

Conversation

Guosmilesmile commented Feb 27, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants