Skip to content

Commit efb2ffe

Browse files
platinumhamburgplatinumhamburg
committed
[FLUSS] Add Producer Offset Snapshot for Exactly-Once semantics
Core changes: - Add ProducerSnapshotManager for lifecycle management with atomic registration - Add ProducerSnapshotStore for ZK + remote storage operations - Add tryRegisterProducerSnapshot in ZooKeeperClient for atomic check-and-create - Add Admin API: registerProducerOffsets, getProducerOffsets, deleteProducerOffsets - Add configurable TTL and cleanup interval for producer snapshots Design highlights: - Atomic registration via ZK's NodeExistsException handling - Eventually consistent: ZK as commit point, orphan files cleaned periodically - UUID-based file naming prevents concurrent upload conflicts Tests: - ProducerSnapshotManagerTest: lifecycle, expiration, concurrent atomicity - ProducerSnapshotJsonSerdeTest: JSON format compatibility
1 parent 864bbe6 commit efb2ffe

18 files changed

Lines changed: 2219 additions & 1 deletion

File tree

‎fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@
7171

7272
import java.util.Collection;
7373
import java.util.List;
74+
import java.util.Map;
7475
import java.util.Optional;
7576
import java.util.concurrent.CompletableFuture;
7677

@@ -604,4 +605,65 @@ CompletableFuture<Optional<RebalanceProgress>> listRebalanceProgress(
604605
* NoRebalanceInProgressException} will be thrown.
605606
*/
606607
CompletableFuture<Void> cancelRebalance(@Nullable String rebalanceId);
608+
609+
// ==================================================================================
610+
// Producer Offset Management APIs (for Exactly-Once Semantics)
611+
// ==================================================================================
612+
613+
/**
614+
* Register producer offset snapshot.
615+
*
616+
* <p>This method provides atomic "check and register" semantics:
617+
*
618+
* <ul>
619+
* <li>If snapshot does not exist: create new snapshot and return {@link
620+
* RegisterResult#CREATED}
621+
* <li>If snapshot already exists: do NOT overwrite and return {@link
622+
* RegisterResult#ALREADY_EXISTS}
623+
* </ul>
624+
*
625+
* <p>The atomicity is guaranteed by the server implementation. This enables the caller to
626+
* determine whether undo recovery is needed based on the return value.
627+
*
628+
* <p>The snapshot will be automatically cleaned up after the configured TTL expires.
629+
*
630+
* <p>This API is typically used by Flink Operator Coordinator at job startup to register the
631+
* initial offset snapshot before any data is written.
632+
*
633+
* @param producerId the ID of the producer (typically Flink job ID)
634+
* @param offsets map of TableBucket to offset for all tables
635+
* @return a CompletableFuture containing the registration result indicating whether the
636+
* snapshot was newly created or already existed
637+
* @since 0.9
638+
*/
639+
CompletableFuture<RegisterResult> registerProducerOffsets(
640+
String producerId, Map<TableBucket, Long> offsets);
641+
642+
/**
643+
* Get producer offset snapshot.
644+
*
645+
* <p>This method retrieves the registered offset snapshot for a producer. Returns null if no
646+
* snapshot exists for the given producer ID.
647+
*
648+
* <p>This API is typically used by Flink Operator Coordinator at job startup to check if a
649+
* previous snapshot exists (indicating a failover before first checkpoint).
650+
*
651+
* @param producerId the ID of the producer
652+
* @return a CompletableFuture containing the producer offsets, or null if not found
653+
* @since 0.9
654+
*/
655+
CompletableFuture<ProducerOffsetsResult> getProducerOffsets(String producerId);
656+
657+
/**
658+
* Delete producer offset snapshot.
659+
*
660+
* <p>This method deletes the registered offset snapshot for a producer. This is typically
661+
* called after the first checkpoint completes successfully, as the checkpoint state will be
662+
* used for recovery instead of the initial snapshot.
663+
*
664+
* @param producerId the ID of the producer
665+
* @return a CompletableFuture that completes when deletion succeeds
666+
* @since 0.9
667+
*/
668+
CompletableFuture<Void> deleteProducerOffsets(String producerId);
607669
}

‎fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@
5757
import org.apache.fluss.rpc.messages.CreateTableRequest;
5858
import org.apache.fluss.rpc.messages.DatabaseExistsRequest;
5959
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
60+
import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest;
6061
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
6162
import org.apache.fluss.rpc.messages.DropAclsRequest;
6263
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
@@ -65,6 +66,7 @@
6566
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest;
6667
import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsRequest;
6768
import org.apache.fluss.rpc.messages.GetLatestLakeSnapshotRequest;
69+
import org.apache.fluss.rpc.messages.GetProducerOffsetsRequest;
6870
import org.apache.fluss.rpc.messages.GetTableInfoRequest;
6971
import org.apache.fluss.rpc.messages.GetTableSchemaRequest;
7072
import org.apache.fluss.rpc.messages.ListAclsRequest;
@@ -76,11 +78,15 @@
7678
import org.apache.fluss.rpc.messages.ListTablesRequest;
7779
import org.apache.fluss.rpc.messages.ListTablesResponse;
7880
import org.apache.fluss.rpc.messages.PbAlterConfig;
81+
import org.apache.fluss.rpc.messages.PbBucketOffset;
7982
import org.apache.fluss.rpc.messages.PbListOffsetsRespForBucket;
8083
import org.apache.fluss.rpc.messages.PbPartitionSpec;
84+
import org.apache.fluss.rpc.messages.PbTableBucketOffset;
85+
import org.apache.fluss.rpc.messages.PbTableOffsets;
8186
import org.apache.fluss.rpc.messages.PbTablePath;
8287
import org.apache.fluss.rpc.messages.RebalanceRequest;
8388
import org.apache.fluss.rpc.messages.RebalanceResponse;
89+
import org.apache.fluss.rpc.messages.RegisterProducerOffsetsRequest;
8490
import org.apache.fluss.rpc.messages.RemoveServerTagRequest;
8591
import org.apache.fluss.rpc.messages.TableExistsRequest;
8692
import org.apache.fluss.rpc.messages.TableExistsResponse;
@@ -588,6 +594,97 @@ public CompletableFuture<Void> cancelRebalance(@Nullable String rebalanceId) {
588594
return gateway.cancelRebalance(request).thenApply(r -> null);
589595
}
590596

597+
// ==================================================================================
598+
// Producer Offset Management APIs (for Exactly-Once Semantics)
599+
// ==================================================================================
600+
601+
@Override
602+
public CompletableFuture<RegisterResult> registerProducerOffsets(
603+
String producerId, Map<TableBucket, Long> offsets) {
604+
checkNotNull(producerId, "producerId must not be null");
605+
checkNotNull(offsets, "offsets must not be null");
606+
607+
RegisterProducerOffsetsRequest request = new RegisterProducerOffsetsRequest();
608+
request.setProducerId(producerId);
609+
610+
// Convert TableBucket offsets to PbTableBucketOffset
611+
for (Map.Entry<TableBucket, Long> entry : offsets.entrySet()) {
612+
TableBucket bucket = entry.getKey();
613+
PbTableBucketOffset pbOffset =
614+
request.addBucketOffset()
615+
.setTableId(bucket.getTableId())
616+
.setBucketId(bucket.getBucket())
617+
.setOffset(entry.getValue());
618+
if (bucket.getPartitionId() != null) {
619+
pbOffset.setPartitionId(bucket.getPartitionId());
620+
}
621+
}
622+
623+
return gateway.registerProducerOffsets(request)
624+
.thenApply(
625+
response -> {
626+
// result: 0 = CREATED, 1 = ALREADY_EXISTS
627+
int result = response.hasResult() ? response.getResult() : 0;
628+
return result == 0
629+
? RegisterResult.CREATED
630+
: RegisterResult.ALREADY_EXISTS;
631+
});
632+
}
633+
634+
@Override
635+
public CompletableFuture<ProducerOffsetsResult> getProducerOffsets(String producerId) {
636+
checkNotNull(producerId, "producerId must not be null");
637+
638+
GetProducerOffsetsRequest request = new GetProducerOffsetsRequest();
639+
request.setProducerId(producerId);
640+
641+
return gateway.getProducerOffsets(request)
642+
.thenApply(
643+
response -> {
644+
if (!response.hasProducerId()) {
645+
// No snapshot found for this producer
646+
return null;
647+
}
648+
649+
Map<Long, Map<TableBucket, Long>> tableOffsets = new HashMap<>();
650+
for (PbTableOffsets pbTableOffsets : response.getTableOffsetsList()) {
651+
long tableId = pbTableOffsets.getTableId();
652+
Map<TableBucket, Long> bucketOffsets = new HashMap<>();
653+
654+
for (PbBucketOffset pbBucketOffset :
655+
pbTableOffsets.getBucketOffsetsList()) {
656+
Long partitionId =
657+
pbBucketOffset.hasPartitionId()
658+
? pbBucketOffset.getPartitionId()
659+
: null;
660+
TableBucket bucket =
661+
new TableBucket(
662+
tableId,
663+
partitionId,
664+
pbBucketOffset.getBucketId());
665+
bucketOffsets.put(bucket, pbBucketOffset.getLogEndOffset());
666+
}
667+
668+
tableOffsets.put(tableId, bucketOffsets);
669+
}
670+
671+
long expirationTime =
672+
response.hasExpirationTime() ? response.getExpirationTime() : 0;
673+
return new ProducerOffsetsResult(
674+
response.getProducerId(), tableOffsets, expirationTime);
675+
});
676+
}
677+
678+
@Override
679+
public CompletableFuture<Void> deleteProducerOffsets(String producerId) {
680+
checkNotNull(producerId, "producerId must not be null");
681+
682+
DeleteProducerOffsetsRequest request = new DeleteProducerOffsetsRequest();
683+
request.setProducerId(producerId);
684+
685+
return gateway.deleteProducerOffsets(request).thenApply(r -> null);
686+
}
687+
591688
@Override
592689
public void close() {
593690
// nothing to do yet

‎fluss-client/src/main/java/org/apache/fluss/client/admin/ProducerOffsetsResult.java‎

Lines changed: 121 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,121 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.fluss.client.admin;
19+
20+
import org.apache.fluss.annotation.PublicEvolving;
21+
import org.apache.fluss.metadata.TableBucket;
22+
23+
import javax.annotation.Nullable;
24+
25+
import java.util.Collections;
26+
import java.util.Map;
27+
28+
/**
29+
* Result containing producer offset snapshot data.
30+
*
31+
* <p>This class holds the offset snapshot for a producer, which is used for undo recovery when a
32+
* Flink job fails over before completing its first checkpoint.
33+
*
34+
* <p>The snapshot contains bucket offsets organized by table ID, allowing the Flink Operator
35+
* Coordinator to coordinate undo recovery across all subtasks.
36+
*
37+
* @since 0.9
38+
*/
39+
@PublicEvolving
40+
public class ProducerOffsetsResult {
41+
42+
private final String producerId;
43+
private final Map<Long, Map<TableBucket, Long>> tableOffsets;
44+
private final long expirationTime;
45+
46+
/**
47+
* Creates a new ProducerOffsetsResult.
48+
*
49+
* @param producerId the producer ID (typically Flink job ID)
50+
* @param tableOffsets map of table ID to bucket offsets
51+
* @param expirationTime the expiration timestamp in milliseconds
52+
*/
53+
public ProducerOffsetsResult(
54+
String producerId,
55+
Map<Long, Map<TableBucket, Long>> tableOffsets,
56+
long expirationTime) {
57+
this.producerId = producerId;
58+
this.tableOffsets = Collections.unmodifiableMap(tableOffsets);
59+
this.expirationTime = expirationTime;
60+
}
61+
62+
/**
63+
* Get the producer ID.
64+
*
65+
* @return the producer ID
66+
*/
67+
public String getProducerId() {
68+
return producerId;
69+
}
70+
71+
/**
72+
* Get the offset snapshot for all tables.
73+
*
74+
* @return unmodifiable map of table ID to bucket offsets
75+
*/
76+
public Map<Long, Map<TableBucket, Long>> getTableOffsets() {
77+
return tableOffsets;
78+
}
79+
80+
/**
81+
* Get the offset snapshot for a specific table.
82+
*
83+
* @param tableId the ID of the table
84+
* @return map of TableBucket to offset, or null if table not found
85+
*/
86+
@Nullable
87+
public Map<TableBucket, Long> getOffsetsForTable(long tableId) {
88+
return tableOffsets.get(tableId);
89+
}
90+
91+
/**
92+
* Get the expiration timestamp.
93+
*
94+
* @return the expiration timestamp in milliseconds since epoch
95+
*/
96+
public long getExpirationTime() {
97+
return expirationTime;
98+
}
99+
100+
/**
101+
* Check if this snapshot has expired.
102+
*
103+
* @return true if the snapshot has expired, false otherwise
104+
*/
105+
public boolean isExpired() {
106+
return System.currentTimeMillis() > expirationTime;
107+
}
108+
109+
@Override
110+
public String toString() {
111+
return "ProducerOffsetsResult{"
112+
+ "producerId='"
113+
+ producerId
114+
+ '\''
115+
+ ", tableCount="
116+
+ tableOffsets.size()
117+
+ ", expirationTime="
118+
+ expirationTime
119+
+ '}';
120+
}
121+
}

‎fluss-client/src/main/java/org/apache/fluss/client/admin/RegisterResult.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.fluss.client.admin;
19+
20+
import org.apache.fluss.annotation.PublicEvolving;
21+
22+
/**
23+
* Result of producer offset registration.
24+
*
25+
* <p>This enum indicates whether a producer offset snapshot was newly created or already existed
26+
* when calling {@link Admin#registerProducerOffsets}.
27+
*
28+
* @since 0.9
29+
*/
30+
@PublicEvolving
31+
public enum RegisterResult {
32+
/**
33+
* Snapshot was newly created.
34+
*
35+
* <p>This indicates a first startup scenario where no previous snapshot existed. The caller
36+
* does not need to perform undo recovery.
37+
*/
38+
CREATED,
39+
40+
/**
41+
* Snapshot already existed and was not overwritten.
42+
*
43+
* <p>This indicates a failover scenario where a previous snapshot exists. The caller should
44+
* perform undo recovery using the existing snapshot offsets.
45+
*/
46+
ALREADY_EXISTS
47+
}

0 commit comments

Comments
 (0)