luoyuxia commented on code in PR #2037:
URL: https://github.com/apache/fluss/pull/2037#discussion_r2581786265
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java:
##########
@@ -1146,50 +1172,52 @@ private <T> void
processAccessContext(AccessContextEvent<T> event) {
}
}
- private CommitLakeTableSnapshotResponse tryProcessCommitLakeTableSnapshot(
- CommitLakeTableSnapshotEvent commitLakeTableSnapshotEvent) {
+ private void tryProcessCommitLakeTableSnapshot(
+ CommitLakeTableSnapshotEvent commitLakeTableSnapshotEvent,
+ CompletableFuture<CommitLakeTableSnapshotResponse> callback) {
+ // commit the lake table snapshot asynchronously
CommitLakeTableSnapshotData commitLakeTableSnapshotData =
commitLakeTableSnapshotEvent.getCommitLakeTableSnapshotData();
- CommitLakeTableSnapshotResponse response = new
CommitLakeTableSnapshotResponse();
Map<Long, LakeTableSnapshot> lakeTableSnapshots =
commitLakeTableSnapshotData.getLakeTableSnapshot();
- for (Map.Entry<Long, LakeTableSnapshot> lakeTableSnapshotEntry :
- lakeTableSnapshots.entrySet()) {
- Long tableId = lakeTableSnapshotEntry.getKey();
-
- PbCommitLakeTableSnapshotRespForTable tableResp =
response.addTableResp();
- tableResp.setTableId(tableId);
-
- try {
- zooKeeperClient.upsertLakeTableSnapshot(tableId,
lakeTableSnapshotEntry.getValue());
- } catch (Exception e) {
- ApiError error = ApiError.fromThrowable(e);
- tableResp.setError(error.error().code(), error.message());
- }
- }
+ ioExecutor.execute(
+ () -> {
+ try {
+ CommitLakeTableSnapshotResponse response =
+ new CommitLakeTableSnapshotResponse();
+ for (Map.Entry<Long, LakeTableSnapshot>
lakeTableSnapshotEntry :
+ lakeTableSnapshots.entrySet()) {
+ Long tableId = lakeTableSnapshotEntry.getKey();
+
+ PbCommitLakeTableSnapshotRespForTable tableResp =
+ response.addTableResp();
+ tableResp.setTableId(tableId);
+
+ try {
+ TablePath tablePath =
coordinatorContext.getTablePathById(tableId);
+ if (tablePath == null) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to find table path
for table id: %d",
+ tableId));
+ }
+ // this involves IO operation (ZK), so we do
it in ioExecutor
+ zooKeeperClient.upsertLakeTableSnapshot(
+ tableId, tablePath,
lakeTableSnapshotEntry.getValue());
+ } catch (Exception e) {
+ ApiError error = ApiError.fromThrowable(e);
+ tableResp.setError(error.error().code(),
error.message());
+ }
+ }
- // send notify lakehouse data request to all replicas.
- coordinatorRequestBatch.newBatch();
- for (Map.Entry<Long, LakeTableSnapshot> lakeTableSnapshotEntry :
- lakeTableSnapshots.entrySet()) {
- LakeTableSnapshot lakeTableSnapshot =
lakeTableSnapshotEntry.getValue();
- for (Map.Entry<TableBucket, Long> bucketLogEndOffsetEntry :
- lakeTableSnapshot.getBucketLogEndOffset().entrySet()) {
- TableBucket tb = bucketLogEndOffsetEntry.getKey();
- coordinatorContext
-
.getBucketLeaderAndIsr(bucketLogEndOffsetEntry.getKey())
- .ifPresent(
- leaderAndIsr ->
- coordinatorRequestBatch
-
.addNotifyLakeTableOffsetRequestForTableServers(
-
coordinatorContext.getAssignment(tb),
- tb,
- lakeTableSnapshot));
- }
- }
- coordinatorRequestBatch.sendNotifyLakeTableOffsetRequest(
- coordinatorContext.getCoordinatorEpoch());
- return response;
+ // send notify lakehouse data request to all replicas
via coordinator event
+ coordinatorEventManager.put(
+ new
NotifyLakeTableOffsetEvent(lakeTableSnapshots));
Review Comment:
Ok.However, Iceberg's snapshot IDs are not strictly incremental, use
snapshot won't work for iceberg. Maybe we can use our internal tiering epoch to
do the check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
