[lake] Record a file path storing log offsets in lake snapshot property #2223

luoyuxia · 2025-12-22T07:30:36Z

Purpose

Linked issue: close #2224

Brief change log

In TieringCommitOperator, first prepare commit log offsets to fluss cluster which will write a file to store the log offsets
then, record the file path in lake snapshot property, commit the snapshot

Tests

Existing test

API and Format

Documentation

Copilot

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

Copilot

Pull request overview

Copilot reviewed 30 out of 30 changed files in this pull request and generated 12 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

...ink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java

fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java

fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java

.../src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java

fluss-common/src/main/java/org/apache/fluss/lake/committer/CommittedLakeSnapshot.java

fluss-rpc/src/main/proto/FlussApi.proto

Copilot

Pull request overview

Copilot reviewed 42 out of 42 changed files in this pull request and generated 3 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerde.java

.../src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java

fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableHelper.java

Copilot

Pull request overview

Copilot reviewed 43 out of 43 changed files in this pull request and generated 6 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

...ink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java

fluss-server/src/main/java/org/apache/fluss/server/entity/CommitLakeTableSnapshotData.java

fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

...common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java

fluss-common/src/test/java/org/apache/fluss/utils/json/TableBucketOffsetsJsonSerdeTest.java

luoyuxia · 2025-12-25T02:29:09Z

@wuchong Could you please help review this pr? The pr also handle the back compabitlity when use v2 to serialize lake table snapshot

fluss-rpc/src/main/proto/FlussApi.proto

wuchong · 2025-12-26T09:14:36Z

fluss-rpc/src/main/proto/FlussApi.proto

+  optional int64 max_timestamp = 6;
+}
+
+message PbPrepareCommitLakeTableRespForTable {


add a table_id field, as the PrepareCommitLakeTableSnapshotRequest request has multiple table ids, we need to distinguish which table is the PbPrepareCommitLakeTableRespForTable belong to.

I was thinking to use array index to distinguish the PbPrepareCommitLakeTableRespForTable belong to. It's a week contract. I'm fine with attaching table id which is a stronger contract, and also align with CommitLakeTableSnapshotRequest

fluss-rpc/src/main/proto/FlussApi.proto

fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTable.java

fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableJsonSerde.java

fluss-server/src/main/java/org/apache/fluss/server/zk/data/lake/LakeTableSnapshotJsonSerde.java

fluss-server/src/test/java/org/apache/fluss/server/zk/data/LakeTableSnapshotJsonSerdeTest.java

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java

luoyuxia · 2025-12-29T06:31:20Z

@wuchong Thanks for review. Comments has been addressed

wuchong

Thanks @luoyuxia for the updating. I only left some final comments.

wuchong · 2025-12-29T09:47:15Z

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

+                                coordinatorEventManager.put(
+                                        new NotifyLakeTableOffsetEvent(
+                                                commitLakeTableSnapshotData.getLakeTableSnapshot(),
+                                                commitLakeTableSnapshotData
+                                                        .getTableBucketsMaxTieredTimestamp()));


Moving the event-publishing logic into the forEach loop would cause the event to be sent multiple times, since the event is intended for all table ids, not per individual table.

wuchong · 2025-12-29T09:47:57Z

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

+                                coordinatorEventManager.put(
+                                        new NotifyLakeTableOffsetEvent(
+                                                lakeTableSnapshots,
+                                                commitLakeTableSnapshotData
+                                                        .getTableBucketsMaxTieredTimestamp()));


wuchong · 2025-12-29T10:03:26Z

fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java

    LIST_REBALANCE_PROGRESS(1050, 0, 0, PUBLIC),
-    CANCEL_REBALANCE(1051, 0, 0, PUBLIC);
+    CANCEL_REBALANCE(1051, 0, 0, PUBLIC),
+    PRE_LAKE_TABLE_SNAPSHOT(1052, 0, 0, PRIVATE);


PRE_LAKE_TABLE_SNAPSHOT -> PREPARE_LAKE_TABLE_SNAPSHOT

While PRE might imply "before," it’s ambiguous and non-idiomatic in event or phase naming. PREPARE clearly conveys the intent.

wuchong · 2025-12-29T10:03:52Z

fluss-rpc/src/main/proto/FlussApi.proto

+  optional int32 error_code = 1;
+  optional string error_message = 2;
+  optional int64 table_id = 3;
+  optional string lake_table_bucket_offsets_path = 4;


lake_table_offsets_path to align with the TableOffsets concept.

wuchong · 2025-12-29T10:04:28Z

fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/CoordinatorGateway.java

+     * Prepares lake table snapshots by merging them with existing snapshots and storing them to the
+     * file system.


Suggested change

* Prepares lake table snapshots by merging them with existing snapshots and storing them to the

* file system.

* Prepares lake table snapshots by merging them with existing snapshots and storing them to the

* remote file system.

wuchong · 2025-12-29T10:04:53Z

fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/CoordinatorGateway.java

+     *       completeness
+     *   <li>Stores the merged snapshot to the remote file system. The stored file contains the log
+     *       end offset information for each bucket in the table
+     *   <li>Returns the file path where the snapshot is stored


Add an item for the second phase.

<li>Call {@link #commitLakeTableSnapshot(CommitLakeTableSnapshotRequest)} with the offset * file path to finalize the snapshot commit to ZooKeeper in the second phase.

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from 33969e9 to 5442462 Compare December 22, 2025 08:45

luoyuxia requested a review from Copilot December 22, 2025 08:47

Copilot AI reviewed Dec 22, 2025

View reviewed changes

Copilot started reviewing on behalf of luoyuxia December 22, 2025 09:19 View session

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from 5442462 to 68a5039 Compare December 22, 2025 12:49

luoyuxia marked this pull request as ready for review December 22, 2025 13:00

luoyuxia requested a review from Copilot December 22, 2025 13:00

Copilot started reviewing on behalf of luoyuxia December 22, 2025 13:01 View session

Copilot AI reviewed Dec 22, 2025

View reviewed changes

luoyuxia force-pushed the allow-commit-offset-to-fluss branch 2 times, most recently from 06af001 to 3889069 Compare December 22, 2025 13:22

luoyuxia changed the title ~~Allow commit offset to fluss~~ [lake] Record a file path storing log offsets in lake snapshot property Dec 23, 2025

luoyuxia mentioned this pull request Dec 24, 2025

[lake] Fix zk lake snapshot node compatible issue #2228

Closed

luoyuxia marked this pull request as draft December 24, 2025 03:07

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from a6c7f69 to 747e91b Compare December 24, 2025 12:35

luoyuxia requested a review from Copilot December 24, 2025 12:36

Copilot started reviewing on behalf of luoyuxia December 24, 2025 12:36 View session

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from 747e91b to f31d6f5 Compare December 24, 2025 12:40

Copilot AI reviewed Dec 24, 2025

View reviewed changes

luoyuxia requested a review from Copilot December 24, 2025 23:56

Copilot started reviewing on behalf of luoyuxia December 24, 2025 23:57 View session

Copilot AI reviewed Dec 25, 2025

View reviewed changes

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from 37c8016 to f7a09fa Compare December 25, 2025 02:06

luoyuxia marked this pull request as ready for review December 25, 2025 02:06

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from f7a09fa to f470b32 Compare December 25, 2025 02:27

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from f470b32 to 0451cbf Compare December 26, 2025 02:19

wuchong reviewed Dec 26, 2025

View reviewed changes

wuchong added the priority=blocker label Dec 28, 2025

[lake] Only record lake snapshot file path in lake snapshots

036e1c1

luoyuxia added 4 commits December 29, 2025 11:14

minor fix

2128d14

refactor

1da58b4

minor fix

68949e1

move prepare commit to coordinator gateway

716b3e0

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from 1c1722e to 9f517ed Compare December 29, 2025 03:25

address comments

97849a9

luoyuxia force-pushed the allow-commit-offset-to-fluss branch from 9f517ed to 97849a9 Compare December 29, 2025 04:02

wuchong reviewed Dec 29, 2025

View reviewed changes

		* Prepares lake table snapshots by merging them with existing snapshots and storing them to the
		* file system.

[lake] Record a file path storing log offsets in lake snapshot property #2223

Are you sure you want to change the base?

[lake] Record a file path storing log offsets in lake snapshot property #2223

Conversation

luoyuxia commented Dec 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

luoyuxia commented Dec 25, 2025

Uh oh!

Uh oh!

Uh oh!

wuchong Dec 26, 2025

Choose a reason for hiding this comment

Uh oh!

luoyuxia Dec 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

luoyuxia commented Dec 29, 2025

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

wuchong Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

wuchong Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

wuchong Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

wuchong Dec 29, 2025

Choose a reason for hiding this comment

luoyuxia commented Dec 22, 2025 •

edited

Loading

luoyuxia Dec 29, 2025 •

edited

Loading