You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/01/27 11:23:48 UTC
(fluss) branch main updated: [kv] Add Producer Offset Snapshot for Exactly-Once semantics (#2434)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 81e3f3a54 [kv] Add Producer Offset Snapshot for Exactly-Once semantics (#2434)
81e3f3a54 is described below
commit 81e3f3a54ca0b7213bf631415da4fc110f6d65ef
Author: Yang Wang <oc...@alibaba-inc.com>
AuthorDate: Tue Jan 27 19:23:43 2026 +0800
[kv] Add Producer Offset Snapshot for Exactly-Once semantics (#2434)
---
.../java/org/apache/fluss/client/admin/Admin.java | 62 +++
.../org/apache/fluss/client/admin/FlussAdmin.java | 46 ++
.../fluss/client/admin/ProducerOffsetsResult.java | 99 ++++
.../apache/fluss/client/admin/RegisterResult.java | 69 +++
.../fluss/client/utils/ClientRpcMessageUtils.java | 91 +++
.../fluss/client/admin/ProducerOffsetsITCase.java | 377 +++++++++++++
.../security/acl/FlussAuthorizationITCase.java | 82 +++
.../org/apache/fluss/config/ConfigOptions.java | 22 +
.../exception/InvalidProducerIdException.java | 41 ++
.../java/org/apache/fluss/utils/FlussPaths.java | 57 ++
.../java/org/apache/fluss/utils/RetryUtils.java | 165 ++++++
.../fluss/utils/function/ThrowingSupplier.java | 37 ++
.../org/apache/fluss/utils/RetryUtilsTest.java | 239 ++++++++
.../org/apache/fluss/rpc/gateway/AdminGateway.java | 40 ++
.../org/apache/fluss/rpc/protocol/ApiKeys.java | 5 +-
.../java/org/apache/fluss/rpc/protocol/Errors.java | 7 +-
fluss-rpc/src/main/proto/FlussApi.proto | 41 ++
.../server/coordinator/CoordinatorService.java | 223 +++++++-
.../producer/ProducerOffsetsManager.java | 611 +++++++++++++++++++++
.../coordinator/producer/ProducerOffsetsStore.java | 390 +++++++++++++
.../fluss/server/utils/ServerRpcMessageUtils.java | 89 +++
.../apache/fluss/server/zk/ZooKeeperClient.java | 134 +++++
.../org/apache/fluss/server/zk/data/ZkData.java | 66 +++
.../server/zk/data/producer/ProducerOffsets.java | 182 ++++++
.../zk/data/producer/ProducerOffsetsJsonSerde.java | 111 ++++
.../server/coordinator/TestCoordinatorGateway.java | 24 +
.../producer/ProducerOffsetsManagerTest.java | 438 +++++++++++++++
.../producer/ProducerOffsetsStoreTest.java | 329 +++++++++++
.../producer/ProducerOffsetsJsonSerdeTest.java | 61 ++
website/docs/maintenance/configuration.md | 8 +-
30 files changed, 4124 insertions(+), 22 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
index 0d8a68cb2..6c0d6876e 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java
@@ -71,6 +71,7 @@ import javax.annotation.Nullable;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -604,4 +605,65 @@ public interface Admin extends AutoCloseable {
* NoRebalanceInProgressException} will be thrown.
*/
CompletableFuture<Void> cancelRebalance(@Nullable String rebalanceId);
+
+ // ==================================================================================
+ // Producer Offset Management APIs (for Exactly-Once Semantics)
+ // ==================================================================================
+
+ /**
+ * Register producer offset snapshot.
+ *
+ * <p>This method provides atomic "check and register" semantics:
+ *
+ * <ul>
+ * <li>If snapshot does not exist: create new snapshot and return {@link
+ * RegisterResult#CREATED}
+ * <li>If snapshot already exists: do NOT overwrite and return {@link
+ * RegisterResult#ALREADY_EXISTS}
+ * </ul>
+ *
+ * <p>The atomicity is guaranteed by the server implementation. This enables the caller to
+ * determine whether undo recovery is needed based on the return value.
+ *
+ * <p>The snapshot will be automatically cleaned up after the configured TTL expires.
+ *
+ * <p>This API is typically used by Flink Operator Coordinator at job startup to register the
+ * initial offset snapshot before any data is written.
+ *
+ * @param producerId the ID of the producer (typically Flink job ID)
+ * @param offsets map of TableBucket to offset for all tables
+ * @return a CompletableFuture containing the registration result indicating whether the
+ * snapshot was newly created or already existed
+ * @since 0.9
+ */
+ CompletableFuture<RegisterResult> registerProducerOffsets(
+ String producerId, Map<TableBucket, Long> offsets);
+
+ /**
+ * Get producer offset snapshot.
+ *
+ * <p>This method retrieves the registered offset snapshot for a producer. Returns null if no
+ * snapshot exists for the given producer ID.
+ *
+ * <p>This API is typically used by Flink Operator Coordinator at job startup to check if a
+ * previous snapshot exists (indicating a failover before first checkpoint).
+ *
+ * @param producerId the ID of the producer
+ * @return a CompletableFuture containing the producer offsets, or null if not found
+ * @since 0.9
+ */
+ CompletableFuture<ProducerOffsetsResult> getProducerOffsets(String producerId);
+
+ /**
+ * Delete producer offset snapshot.
+ *
+ * <p>This method deletes the registered offset snapshot for a producer. This is typically
+ * called after the first checkpoint completes successfully, as the checkpoint state will be
+ * used for recovery instead of the initial snapshot.
+ *
+ * @param producerId the ID of the producer
+ * @return a CompletableFuture that completes when deletion succeeds
+ * @since 0.9
+ */
+ CompletableFuture<Void> deleteProducerOffsets(String producerId);
}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
index 88aec06c8..543d65128 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
@@ -57,6 +57,7 @@ import org.apache.fluss.rpc.messages.CreateDatabaseRequest;
import org.apache.fluss.rpc.messages.CreateTableRequest;
import org.apache.fluss.rpc.messages.DatabaseExistsRequest;
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
+import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
import org.apache.fluss.rpc.messages.DropAclsRequest;
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
@@ -65,6 +66,7 @@ import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest;
import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsRequest;
import org.apache.fluss.rpc.messages.GetLatestLakeSnapshotRequest;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsRequest;
import org.apache.fluss.rpc.messages.GetTableInfoRequest;
import org.apache.fluss.rpc.messages.GetTableSchemaRequest;
import org.apache.fluss.rpc.messages.ListAclsRequest;
@@ -105,6 +107,7 @@ import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeCreatePart
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeDropPartitionRequest;
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeListOffsetsRequest;
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makePbPartitionSpec;
+import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeRegisterProducerOffsetsRequest;
import static org.apache.fluss.client.utils.ClientRpcMessageUtils.toConfigEntries;
import static org.apache.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclBindings;
@@ -588,6 +591,49 @@ public class FlussAdmin implements Admin {
return gateway.cancelRebalance(request).thenApply(r -> null);
}
+ // ==================================================================================
+ // Producer Offset Management APIs (for Exactly-Once Semantics)
+ // ==================================================================================
+
+ @Override
+ public CompletableFuture<RegisterResult> registerProducerOffsets(
+ String producerId, Map<TableBucket, Long> offsets) {
+ checkNotNull(producerId, "producerId must not be null");
+ checkNotNull(offsets, "offsets must not be null");
+
+ return gateway.registerProducerOffsets(
+ makeRegisterProducerOffsetsRequest(producerId, offsets))
+ .thenApply(
+ response -> {
+ int code =
+ response.hasResult()
+ ? response.getResult()
+ : RegisterResult.CREATED.getCode();
+ return RegisterResult.fromCode(code);
+ });
+ }
+
+ @Override
+ public CompletableFuture<ProducerOffsetsResult> getProducerOffsets(String producerId) {
+ checkNotNull(producerId, "producerId must not be null");
+
+ GetProducerOffsetsRequest request = new GetProducerOffsetsRequest();
+ request.setProducerId(producerId);
+
+ return gateway.getProducerOffsets(request)
+ .thenApply(ClientRpcMessageUtils::toProducerOffsetsResult);
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteProducerOffsets(String producerId) {
+ checkNotNull(producerId, "producerId must not be null");
+
+ DeleteProducerOffsetsRequest request = new DeleteProducerOffsetsRequest();
+ request.setProducerId(producerId);
+
+ return gateway.deleteProducerOffsets(request).thenApply(r -> null);
+ }
+
@Override
public void close() {
// nothing to do yet
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/ProducerOffsetsResult.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/ProducerOffsetsResult.java
new file mode 100644
index 000000000..7f48eb905
--- /dev/null
+++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/ProducerOffsetsResult.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.admin;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.metadata.TableBucket;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Result containing producer offset snapshot data.
+ *
+ * <p>This class holds the offset snapshot for a producer, which is used for undo recovery when a
+ * Flink job fails over before completing its first checkpoint.
+ *
+ * <p>The snapshot contains bucket offsets organized by table ID, allowing the Flink Operator
+ * Coordinator to coordinate undo recovery across all subtasks.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class ProducerOffsetsResult {
+
+ private final String producerId;
+ private final Map<Long, Map<TableBucket, Long>> tableOffsets;
+ private final long expirationTime;
+
+ /**
+ * Creates a new ProducerOffsetsResult.
+ *
+ * @param producerId the producer ID (typically Flink job ID)
+ * @param tableOffsets map of table ID to bucket offsets
+ * @param expirationTime the expiration timestamp in milliseconds
+ */
+ public ProducerOffsetsResult(
+ String producerId,
+ Map<Long, Map<TableBucket, Long>> tableOffsets,
+ long expirationTime) {
+ this.producerId = producerId;
+ this.tableOffsets = Collections.unmodifiableMap(tableOffsets);
+ this.expirationTime = expirationTime;
+ }
+
+ /**
+ * Get the producer ID.
+ *
+ * @return the producer ID
+ */
+ public String getProducerId() {
+ return producerId;
+ }
+
+ /**
+ * Get the offset snapshot for all tables.
+ *
+ * @return unmodifiable map of table ID to bucket offsets
+ */
+ public Map<Long, Map<TableBucket, Long>> getTableOffsets() {
+ return tableOffsets;
+ }
+
+ /**
+ * Get the expiration timestamp.
+ *
+ * @return the expiration timestamp in milliseconds since epoch
+ */
+ public long getExpirationTime() {
+ return expirationTime;
+ }
+
+ @Override
+ public String toString() {
+ return "ProducerOffsetsResult{"
+ + "producerId='"
+ + producerId
+ + '\''
+ + ", tableCount="
+ + tableOffsets.size()
+ + ", expirationTime="
+ + expirationTime
+ + '}';
+ }
+}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/RegisterResult.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/RegisterResult.java
new file mode 100644
index 000000000..1d373bf0e
--- /dev/null
+++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/RegisterResult.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.admin;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * Result of producer offset registration.
+ *
+ * <p>This enum indicates whether a producer offset snapshot was newly created or already existed
+ * when calling registerProducerOffsets API.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public enum RegisterResult {
+ /**
+ * Snapshot was newly created.
+ *
+ * <p>This indicates a first startup scenario where no previous snapshot existed. The caller
+ * does not need to perform undo recovery.
+ */
+ CREATED(0),
+
+ /**
+ * Snapshot already existed and was not overwritten.
+ *
+ * <p>This indicates a failover scenario where a previous snapshot exists. The caller should
+ * perform undo recovery using the existing snapshot offsets.
+ */
+ ALREADY_EXISTS(1);
+
+ /** The code used in RPC messages. */
+ private final int code;
+
+ RegisterResult(int code) {
+ this.code = code;
+ }
+
+ /** Returns the code used in RPC messages. */
+ public int getCode() {
+ return code;
+ }
+
+ /** Returns the RegisterResult for the given code. */
+ public static RegisterResult fromCode(int code) {
+ for (RegisterResult result : values()) {
+ if (result.code == code) {
+ return result;
+ }
+ }
+ throw new IllegalArgumentException("Unknown RegisterResult code: " + code);
+ }
+}
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
index 2fae90d4b..f6b37f5d1 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java
@@ -18,6 +18,7 @@
package org.apache.fluss.client.utils;
import org.apache.fluss.client.admin.OffsetSpec;
+import org.apache.fluss.client.admin.ProducerOffsetsResult;
import org.apache.fluss.client.lookup.LookupBatch;
import org.apache.fluss.client.lookup.PrefixLookupBatch;
import org.apache.fluss.client.metadata.KvSnapshotMetadata;
@@ -48,6 +49,7 @@ import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse;
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataResponse;
import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsResponse;
import org.apache.fluss.rpc.messages.GetLatestLakeSnapshotResponse;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.ListOffsetsRequest;
import org.apache.fluss.rpc.messages.ListPartitionInfosResponse;
import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse;
@@ -55,6 +57,7 @@ import org.apache.fluss.rpc.messages.LookupRequest;
import org.apache.fluss.rpc.messages.MetadataRequest;
import org.apache.fluss.rpc.messages.PbAddColumn;
import org.apache.fluss.rpc.messages.PbAlterConfig;
+import org.apache.fluss.rpc.messages.PbBucketOffset;
import org.apache.fluss.rpc.messages.PbDescribeConfig;
import org.apache.fluss.rpc.messages.PbDropColumn;
import org.apache.fluss.rpc.messages.PbKeyValue;
@@ -65,6 +68,7 @@ import org.apache.fluss.rpc.messages.PbModifyColumn;
import org.apache.fluss.rpc.messages.PbPartitionSpec;
import org.apache.fluss.rpc.messages.PbPrefixLookupReqForBucket;
import org.apache.fluss.rpc.messages.PbProduceLogReqForBucket;
+import org.apache.fluss.rpc.messages.PbProducerTableOffsets;
import org.apache.fluss.rpc.messages.PbPutKvReqForBucket;
import org.apache.fluss.rpc.messages.PbRebalancePlanForBucket;
import org.apache.fluss.rpc.messages.PbRebalanceProgressForBucket;
@@ -74,6 +78,7 @@ import org.apache.fluss.rpc.messages.PbRenameColumn;
import org.apache.fluss.rpc.messages.PrefixLookupRequest;
import org.apache.fluss.rpc.messages.ProduceLogRequest;
import org.apache.fluss.rpc.messages.PutKvRequest;
+import org.apache.fluss.rpc.messages.RegisterProducerOffsetsRequest;
import org.apache.fluss.utils.json.DataTypeJsonSerde;
import org.apache.fluss.utils.json.JsonSerdeUtils;
@@ -542,4 +547,90 @@ public class ClientRpcMessageUtils {
pbDescribeConfig.getConfigSource())))
.collect(Collectors.toList());
}
+
+ /**
+ * Parses a PbProducerTableOffsets into a map of TableBucket to offset.
+ *
+ * @param pbTableOffsets the protobuf producer table offsets
+ * @return map of TableBucket to offset
+ */
+ public static Map<TableBucket, Long> toTableBucketOffsets(
+ PbProducerTableOffsets pbTableOffsets) {
+ Map<TableBucket, Long> bucketOffsets = new HashMap<>();
+ long tableId = pbTableOffsets.getTableId();
+
+ for (PbBucketOffset pbBucketOffset : pbTableOffsets.getBucketOffsetsList()) {
+ Long partitionId =
+ pbBucketOffset.hasPartitionId() ? pbBucketOffset.getPartitionId() : null;
+ TableBucket bucket =
+ new TableBucket(tableId, partitionId, pbBucketOffset.getBucketId());
+ bucketOffsets.put(bucket, pbBucketOffset.getLogEndOffset());
+ }
+
+ return bucketOffsets;
+ }
+
+ /**
+ * Creates a RegisterProducerOffsetsRequest from producer ID and offsets map.
+ *
+ * @param producerId the producer ID
+ * @param offsets map of TableBucket to offset
+ * @return the RegisterProducerOffsetsRequest
+ */
+ public static RegisterProducerOffsetsRequest makeRegisterProducerOffsetsRequest(
+ String producerId, Map<TableBucket, Long> offsets) {
+ RegisterProducerOffsetsRequest request = new RegisterProducerOffsetsRequest();
+ request.setProducerId(producerId);
+
+ // Group offsets by table ID
+ Map<Long, List<Map.Entry<TableBucket, Long>>> offsetsByTable = new HashMap<>();
+ for (Map.Entry<TableBucket, Long> entry : offsets.entrySet()) {
+ offsetsByTable
+ .computeIfAbsent(entry.getKey().getTableId(), k -> new ArrayList<>())
+ .add(entry);
+ }
+
+ // Build PbProducerTableOffsets for each table
+ for (Map.Entry<Long, List<Map.Entry<TableBucket, Long>>> tableEntry :
+ offsetsByTable.entrySet()) {
+ PbProducerTableOffsets pbTableOffsets =
+ request.addTableOffset().setTableId(tableEntry.getKey());
+ for (Map.Entry<TableBucket, Long> bucketEntry : tableEntry.getValue()) {
+ TableBucket bucket = bucketEntry.getKey();
+ PbBucketOffset pbBucketOffset =
+ pbTableOffsets
+ .addBucketOffset()
+ .setBucketId(bucket.getBucket())
+ .setLogEndOffset(bucketEntry.getValue());
+ if (bucket.getPartitionId() != null) {
+ pbBucketOffset.setPartitionId(bucket.getPartitionId());
+ }
+ }
+ }
+
+ return request;
+ }
+
+ /**
+ * Converts a GetProducerOffsetsResponse to ProducerOffsetsResult.
+ *
+ * @param response the GetProducerOffsetsResponse
+ * @return the ProducerOffsetsResult, or null if producer not found
+ */
+ @Nullable
+ public static ProducerOffsetsResult toProducerOffsetsResult(
+ GetProducerOffsetsResponse response) {
+ if (!response.hasProducerId()) {
+ return null;
+ }
+
+ Map<Long, Map<TableBucket, Long>> tableOffsets = new HashMap<>();
+ for (PbProducerTableOffsets pbTableOffsets : response.getTableOffsetsList()) {
+ long tableId = pbTableOffsets.getTableId();
+ tableOffsets.put(tableId, toTableBucketOffsets(pbTableOffsets));
+ }
+
+ long expirationTime = response.hasExpirationTime() ? response.getExpirationTime() : 0;
+ return new ProducerOffsetsResult(response.getProducerId(), tableOffsets, expirationTime);
+ }
}
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/ProducerOffsetsITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/ProducerOffsetsITCase.java
new file mode 100644
index 000000000..0227ff71a
--- /dev/null
+++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/ProducerOffsetsITCase.java
@@ -0,0 +1,377 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.admin;
+
+import org.apache.fluss.metadata.TableBucket;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration tests for Producer Offset Snapshot APIs.
+ *
+ * <p>These tests verify the end-to-end functionality of:
+ *
+ * <ul>
+ * <li>{@link Admin#registerProducerOffsets(String, Map)}
+ * <li>{@link Admin#getProducerOffsets(String)}
+ * <li>{@link Admin#deleteProducerOffsets(String)}
+ * </ul>
+ */
+class ProducerOffsetsITCase extends ClientToServerITCaseBase {
+
+ // ------------------------------------------------------------------------
+ // Basic CRUD Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRegisterAndGetProducerOffsets() throws Exception {
+ String producerId = "test-producer-" + System.currentTimeMillis();
+ Map<TableBucket, Long> offsets = createTestOffsets();
+
+ // Register offsets
+ RegisterResult result = admin.registerProducerOffsets(producerId, offsets).get();
+ assertThat(result).isEqualTo(RegisterResult.CREATED);
+
+ // Get offsets
+ ProducerOffsetsResult offsetsResult = admin.getProducerOffsets(producerId).get();
+ assertThat(offsetsResult).isNotNull();
+ assertThat(offsetsResult.getProducerId()).isEqualTo(producerId);
+ assertThat(offsetsResult.getExpirationTime()).isGreaterThan(System.currentTimeMillis());
+
+ // Verify offsets content
+ Map<Long, Map<TableBucket, Long>> tableOffsets = offsetsResult.getTableOffsets();
+ assertThat(tableOffsets).isNotEmpty();
+
+ // Flatten and verify all offsets are present
+ Map<TableBucket, Long> allOffsets = flattenTableOffsets(tableOffsets);
+ assertThat(allOffsets).hasSize(offsets.size());
+ for (Map.Entry<TableBucket, Long> entry : offsets.entrySet()) {
+ assertThat(allOffsets).containsEntry(entry.getKey(), entry.getValue());
+ }
+
+ // Cleanup
+ admin.deleteProducerOffsets(producerId).get();
+ }
+
+ @Test
+ void testRegisterProducerOffsetsAlreadyExists() throws Exception {
+ String producerId = "test-producer-exists-" + System.currentTimeMillis();
+ Map<TableBucket, Long> offsets1 = new HashMap<>();
+ offsets1.put(new TableBucket(1L, 0), 100L);
+
+ Map<TableBucket, Long> offsets2 = new HashMap<>();
+ offsets2.put(new TableBucket(2L, 0), 200L);
+
+ try {
+ // First registration should succeed with CREATED
+ RegisterResult result1 = admin.registerProducerOffsets(producerId, offsets1).get();
+ assertThat(result1).isEqualTo(RegisterResult.CREATED);
+
+ // Second registration should return ALREADY_EXISTS
+ RegisterResult result2 = admin.registerProducerOffsets(producerId, offsets2).get();
+ assertThat(result2).isEqualTo(RegisterResult.ALREADY_EXISTS);
+
+ // Verify original offsets are preserved
+ ProducerOffsetsResult offsetsResult = admin.getProducerOffsets(producerId).get();
+ Map<TableBucket, Long> allOffsets =
+ flattenTableOffsets(offsetsResult.getTableOffsets());
+ assertThat(allOffsets).containsKey(new TableBucket(1L, 0));
+ assertThat(allOffsets).doesNotContainKey(new TableBucket(2L, 0));
+ } finally {
+ admin.deleteProducerOffsets(producerId).get();
+ }
+ }
+
+ @Test
+ void testGetProducerOffsetsNotFound() throws Exception {
+ String producerId = "non-existent-producer-" + System.currentTimeMillis();
+
+ ProducerOffsetsResult result = admin.getProducerOffsets(producerId).get();
+ assertThat(result).isNull();
+ }
+
+ @Test
+ void testDeleteProducerOffsets() throws Exception {
+ String producerId = "test-producer-delete-" + System.currentTimeMillis();
+ Map<TableBucket, Long> offsets = createTestOffsets();
+
+ // Register offsets
+ admin.registerProducerOffsets(producerId, offsets).get();
+ assertThat(admin.getProducerOffsets(producerId).get()).isNotNull();
+
+ // Delete offsets
+ admin.deleteProducerOffsets(producerId).get();
+
+ // Verify deleted
+ assertThat(admin.getProducerOffsets(producerId).get()).isNull();
+ }
+
+ @Test
+ void testDeleteNonExistentProducerOffsets() throws Exception {
+ String producerId = "non-existent-producer-delete-" + System.currentTimeMillis();
+
+ // Should not throw exception
+ admin.deleteProducerOffsets(producerId).get();
+ }
+
+ // ------------------------------------------------------------------------
+ // Partitioned Table Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRegisterOffsetsWithPartitionedTable() throws Exception {
+ String producerId = "test-producer-partitioned-" + System.currentTimeMillis();
+ Map<TableBucket, Long> offsets = new HashMap<>();
+
+ // Non-partitioned table
+ offsets.put(new TableBucket(1L, 0), 100L);
+ offsets.put(new TableBucket(1L, 1), 101L);
+
+ // Partitioned table with different partitions
+ offsets.put(new TableBucket(2L, 10L, 0), 200L);
+ offsets.put(new TableBucket(2L, 10L, 1), 201L);
+ offsets.put(new TableBucket(2L, 20L, 0), 300L);
+ offsets.put(new TableBucket(2L, 20L, 1), 301L);
+
+ try {
+ // Register
+ RegisterResult result = admin.registerProducerOffsets(producerId, offsets).get();
+ assertThat(result).isEqualTo(RegisterResult.CREATED);
+
+ // Get and verify
+ ProducerOffsetsResult offsetsResult = admin.getProducerOffsets(producerId).get();
+ assertThat(offsetsResult).isNotNull();
+
+ Map<TableBucket, Long> allOffsets =
+ flattenTableOffsets(offsetsResult.getTableOffsets());
+ assertThat(allOffsets).hasSize(6);
+
+ // Verify non-partitioned table offsets
+ assertThat(allOffsets.get(new TableBucket(1L, 0))).isEqualTo(100L);
+ assertThat(allOffsets.get(new TableBucket(1L, 1))).isEqualTo(101L);
+
+ // Verify partitioned table offsets
+ assertThat(allOffsets.get(new TableBucket(2L, 10L, 0))).isEqualTo(200L);
+ assertThat(allOffsets.get(new TableBucket(2L, 10L, 1))).isEqualTo(201L);
+ assertThat(allOffsets.get(new TableBucket(2L, 20L, 0))).isEqualTo(300L);
+ assertThat(allOffsets.get(new TableBucket(2L, 20L, 1))).isEqualTo(301L);
+ } finally {
+ admin.deleteProducerOffsets(producerId).get();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Concurrent Registration Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testConcurrentRegistrationSameProducer() throws Exception {
+ String producerId = "test-concurrent-same-" + System.currentTimeMillis();
+ int numThreads = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ CountDownLatch doneLatch = new CountDownLatch(numThreads);
+ AtomicInteger createdCount = new AtomicInteger(0);
+ AtomicInteger existsCount = new AtomicInteger(0);
+
+ try {
+ for (int i = 0; i < numThreads; i++) {
+ final int threadId = i;
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ startLatch.await();
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, threadId), (long) threadId);
+
+ RegisterResult result =
+ admin.registerProducerOffsets(producerId, offsets).get();
+ if (result == RegisterResult.CREATED) {
+ createdCount.incrementAndGet();
+ } else {
+ existsCount.incrementAndGet();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ } finally {
+ doneLatch.countDown();
+ }
+ },
+ executor);
+ }
+
+ // Start all threads simultaneously
+ startLatch.countDown();
+ assertThat(doneLatch.await(30, TimeUnit.SECONDS)).isTrue();
+
+ // Exactly one should succeed with CREATED
+ assertThat(createdCount.get()).isEqualTo(1);
+ assertThat(existsCount.get()).isEqualTo(numThreads - 1);
+
+ // Verify snapshot exists
+ assertThat(admin.getProducerOffsets(producerId).get()).isNotNull();
+ } finally {
+ executor.shutdown();
+ admin.deleteProducerOffsets(producerId).get();
+ }
+ }
+
+ @Test
+ void testConcurrentRegistrationDifferentProducers() throws Exception {
+ String prefix = "test-concurrent-diff-" + System.currentTimeMillis() + "-";
+ int numProducers = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(numProducers);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ CountDownLatch doneLatch = new CountDownLatch(numProducers);
+ AtomicInteger successCount = new AtomicInteger(0);
+ List<String> producerIds = new ArrayList<>();
+
+ try {
+ for (int i = 0; i < numProducers; i++) {
+ final int producerNum = i;
+ final String producerId = prefix + producerNum;
+ producerIds.add(producerId);
+
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ startLatch.await();
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, producerNum), (long) producerNum);
+
+ RegisterResult result =
+ admin.registerProducerOffsets(producerId, offsets).get();
+ if (result == RegisterResult.CREATED) {
+ successCount.incrementAndGet();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ } finally {
+ doneLatch.countDown();
+ }
+ },
+ executor);
+ }
+
+ // Start all threads simultaneously
+ startLatch.countDown();
+ assertThat(doneLatch.await(30, TimeUnit.SECONDS)).isTrue();
+
+ // All should succeed since they're different producers
+ assertThat(successCount.get()).isEqualTo(numProducers);
+
+ // Verify all snapshots exist
+ for (String producerId : producerIds) {
+ assertThat(admin.getProducerOffsets(producerId).get()).isNotNull();
+ }
+ } finally {
+ executor.shutdown();
+ // Cleanup all producers
+ for (String producerId : producerIds) {
+ admin.deleteProducerOffsets(producerId).get();
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Edge Cases
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRegisterEmptyOffsets() throws Exception {
+ String producerId = "test-producer-empty-" + System.currentTimeMillis();
+ Map<TableBucket, Long> offsets = new HashMap<>();
+
+ try {
+ RegisterResult result = admin.registerProducerOffsets(producerId, offsets).get();
+ assertThat(result).isEqualTo(RegisterResult.CREATED);
+
+ ProducerOffsetsResult offsetsResult = admin.getProducerOffsets(producerId).get();
+ assertThat(offsetsResult).isNotNull();
+ assertThat(offsetsResult.getTableOffsets()).isEmpty();
+ } finally {
+ admin.deleteProducerOffsets(producerId).get();
+ }
+ }
+
+ @Test
+ void testRegisterLargeNumberOfOffsets() throws Exception {
+ String producerId = "test-producer-large-" + System.currentTimeMillis();
+ Map<TableBucket, Long> offsets = new HashMap<>();
+
+ // Create offsets for 100 tables with 10 buckets each
+ for (long tableId = 1; tableId <= 100; tableId++) {
+ for (int bucket = 0; bucket < 10; bucket++) {
+ offsets.put(new TableBucket(tableId, bucket), tableId * 1000 + bucket);
+ }
+ }
+
+ try {
+ RegisterResult result = admin.registerProducerOffsets(producerId, offsets).get();
+ assertThat(result).isEqualTo(RegisterResult.CREATED);
+
+ ProducerOffsetsResult offsetsResult = admin.getProducerOffsets(producerId).get();
+ assertThat(offsetsResult).isNotNull();
+
+ Map<TableBucket, Long> allOffsets =
+ flattenTableOffsets(offsetsResult.getTableOffsets());
+ assertThat(allOffsets).hasSize(1000);
+
+ // Verify some random offsets
+ assertThat(allOffsets.get(new TableBucket(1L, 0))).isEqualTo(1000L);
+ assertThat(allOffsets.get(new TableBucket(50L, 5))).isEqualTo(50005L);
+ assertThat(allOffsets.get(new TableBucket(100L, 9))).isEqualTo(100009L);
+ } finally {
+ admin.deleteProducerOffsets(producerId).get();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Helper Methods
+ // ------------------------------------------------------------------------
+
+ private Map<TableBucket, Long> createTestOffsets() {
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+ offsets.put(new TableBucket(1L, 1), 200L);
+ offsets.put(new TableBucket(2L, 0), 300L);
+ return offsets;
+ }
+
+ private Map<TableBucket, Long> flattenTableOffsets(
+ Map<Long, Map<TableBucket, Long>> tableOffsets) {
+ Map<TableBucket, Long> result = new HashMap<>();
+ for (Map<TableBucket, Long> bucketOffsets : tableOffsets.values()) {
+ result.putAll(bucketOffsets);
+ }
+ return result;
+ }
+}
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
index cdb92ab95..89c64e5ca 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java
@@ -83,7 +83,9 @@ import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ExecutionException;
import static org.apache.fluss.config.ConfigOptions.DATALAKE_FORMAT;
@@ -1020,6 +1022,86 @@ public class FlussAuthorizationITCase {
guestAdmin.cancelRebalance(null).get();
}
+ // ------------------------------------------------------------------------
+ // Producer Offsets Authorization Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRegisterProducerOffsetsAuthorization() throws Exception {
+ String producerId = "test-producer-auth-" + System.currentTimeMillis();
+ TableInfo tableInfo = rootAdmin.getTableInfo(DATA1_TABLE_PATH_PK).get();
+ long tableId = tableInfo.getTableId();
+
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(tableId, 0), 100L);
+
+ // test registerProducerOffsets without WRITE permission on table resource
+ assertThatThrownBy(() -> guestAdmin.registerProducerOffsets(producerId, offsets).get())
+ .rootCause()
+ .isInstanceOf(AuthorizationException.class)
+ .hasMessageContaining(
+ String.format(
+ "Principal %s have no authorization to operate WRITE on resource Resource{type=TABLE, name='%s'}",
+ guestPrincipal, DATA1_TABLE_PATH_PK));
+
+ // add WRITE permission to guest user on table resource
+ List<AclBinding> aclBindings =
+ Collections.singletonList(
+ new AclBinding(
+ Resource.table(DATA1_TABLE_PATH_PK),
+ new AccessControlEntry(
+ guestPrincipal,
+ "*",
+ OperationType.WRITE,
+ PermissionType.ALLOW)));
+ rootAdmin.createAcls(aclBindings).all().get();
+ FLUSS_CLUSTER_EXTENSION.waitUntilAuthenticationSync(aclBindings, true);
+
+ // test registerProducerOffsets with WRITE permission should succeed
+ guestAdmin.registerProducerOffsets(producerId, offsets).get();
+
+ // cleanup
+ rootAdmin.deleteProducerOffsets(producerId).get();
+ }
+
+ @Test
+ void testDeleteProducerOffsetsAuthorization() throws Exception {
+ String producerId = "test-producer-delete-auth-" + System.currentTimeMillis();
+ TableInfo tableInfo = rootAdmin.getTableInfo(DATA1_TABLE_PATH_PK).get();
+ long tableId = tableInfo.getTableId();
+
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(tableId, 0), 100L);
+
+ // register offsets as root user first
+ rootAdmin.registerProducerOffsets(producerId, offsets).get();
+
+ // test deleteProducerOffsets without WRITE permission on table resource
+ assertThatThrownBy(() -> guestAdmin.deleteProducerOffsets(producerId).get())
+ .rootCause()
+ .isInstanceOf(AuthorizationException.class)
+ .hasMessageContaining(
+ String.format(
+ "Principal %s have no authorization to operate WRITE on resource Resource{type=TABLE, name='%s'}",
+ guestPrincipal, DATA1_TABLE_PATH_PK));
+
+ // add WRITE permission to guest user on table resource
+ List<AclBinding> aclBindings =
+ Collections.singletonList(
+ new AclBinding(
+ Resource.table(DATA1_TABLE_PATH_PK),
+ new AccessControlEntry(
+ guestPrincipal,
+ "*",
+ OperationType.WRITE,
+ PermissionType.ALLOW)));
+ rootAdmin.createAcls(aclBindings).all().get();
+ FLUSS_CLUSTER_EXTENSION.waitUntilAuthenticationSync(aclBindings, true);
+
+ // test deleteProducerOffsets with WRITE permission should succeed
+ guestAdmin.deleteProducerOffsets(producerId).get();
+ }
+
private static Configuration initConfig() {
Configuration conf = new Configuration();
conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 3);
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
index 7e04c3a1c..e7dbec436 100644
--- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
+++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -354,6 +354,28 @@ public class ConfigOptions {
+ "The default value is 10. "
+ "This option is deprecated. Please use server.io-pool.size instead.");
+ /**
+ * The TTL (time-to-live) for producer offsets. Producer offsets older than this TTL will be
+ * automatically cleaned up by the coordinator server.
+ */
+ public static final ConfigOption<Duration> COORDINATOR_PRODUCER_OFFSETS_TTL =
+ key("coordinator.producer-offsets.ttl")
+ .durationType()
+ .defaultValue(Duration.ofHours(24))
+ .withDescription(
+ "The TTL (time-to-live) for producer offsets. "
+ + "Producer offsets older than this TTL will be automatically cleaned up "
+ + "by the coordinator server. Default is 24 hours.");
+
+ /** The interval for cleaning up expired producer offsets and orphan files in remote storage. */
+ public static final ConfigOption<Duration> COORDINATOR_PRODUCER_OFFSETS_CLEANUP_INTERVAL =
+ key("coordinator.producer-offsets.cleanup-interval")
+ .durationType()
+ .defaultValue(Duration.ofHours(1))
+ .withDescription(
+ "The interval for cleaning up expired producer offsets "
+ + "and orphan files in remote storage. Default is 1 hour.");
+
// ------------------------------------------------------------------------
// ConfigOptions for Tablet Server
// ------------------------------------------------------------------------
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/InvalidProducerIdException.java b/fluss-common/src/main/java/org/apache/fluss/exception/InvalidProducerIdException.java
new file mode 100644
index 000000000..14c3b357b
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/InvalidProducerIdException.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.exception;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * The client has attempted to perform an operation with an invalid producer ID. For example the
+ * producer ID is null, empty, too long, or contains invalid characters. This exception is not
+ * retriable because the operation won't suddenly become valid.
+ *
+ * @since 0.9
+ */
+@PublicEvolving
+public class InvalidProducerIdException extends ApiException {
+
+ private static final long serialVersionUID = 1L;
+
+ public InvalidProducerIdException(String message) {
+ super(message);
+ }
+
+ public InvalidProducerIdException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java b/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java
index d04268fba..c2e796964 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/FlussPaths.java
@@ -93,6 +93,12 @@ public class FlussPaths {
private static final String REMOTE_LAKE_DIR_NAME = "lake";
+ /** The directory name for storing producer offsets files. */
+ private static final String REMOTE_PRODUCERS_DIR_NAME = "producers";
+
+ /** Suffix of a producer offsets file. */
+ private static final String PRODUCER_OFFSETS_FILE_SUFFIX = ".offsets";
+
// ----------------------------------------------------------------------------------------
// LOG/KV Tablet Paths
// ----------------------------------------------------------------------------------------
@@ -739,6 +745,57 @@ public class FlussPaths {
return new FsPath(remoteKvTabletDir, REMOTE_KV_SNAPSHOT_SHARED_DIR);
}
+ // ----------------------------------------------------------------------------------------
+ // Remote Producer Offsets Paths
+ // ----------------------------------------------------------------------------------------
+
+ /**
+ * Returns the remote root directory path for storing producer offsets files.
+ *
+ * <p>The path contract:
+ *
+ * <pre>
+ * {$remote.data.dir}/producers
+ * </pre>
+ *
+ * @param remoteDataDir the remote data root directory, i.e. the "remote.data.dir" in the
+ * configuration
+ */
+ public static FsPath remoteProducersDir(String remoteDataDir) {
+ return new FsPath(remoteDataDir, REMOTE_PRODUCERS_DIR_NAME);
+ }
+
+ /**
+ * Returns the remote file path for storing producer offsets for a specific table.
+ *
+ * <p>Producer offsets are stored per producer and per table. Each file contains the offsets for
+ * all buckets of a single table that the producer has written to. The file is named with a UUID
+ * to ensure uniqueness and avoid conflicts during concurrent writes.
+ *
+ * <p>The path contract:
+ *
+ * <pre>
+ * {$remote.data.dir}/producers/{producerId}/{tableId}/{uuid}.offsets
+ * </pre>
+ *
+ * <p>For example: {@code s3://bucket/fluss/producers/my-producer-1/12345/550e8400-e29b.offsets}
+ *
+ * @param remoteDataDir the remote data root directory, i.e. the "remote.data.dir" in the
+ * configuration
+ * @param producerId the unique identifier of the producer (e.g., Flink job ID)
+ * @param tableId the table ID for which offsets are being stored
+ * @param uuid a unique identifier for this specific offsets file
+ * @return the full path to the producer offsets file
+ */
+ public static FsPath remoteProducerOffsetsPath(
+ String remoteDataDir, String producerId, long tableId, UUID uuid) {
+ return new FsPath(
+ new FsPath(
+ new FsPath(remoteProducersDir(remoteDataDir), producerId),
+ String.valueOf(tableId)),
+ uuid + PRODUCER_OFFSETS_FILE_SUFFIX);
+ }
+
// ----------------------------------------------------------------------------------------
// Remote Paths for common use
// ----------------------------------------------------------------------------------------
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/RetryUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/RetryUtils.java
new file mode 100644
index 000000000..dba1422a9
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/RetryUtils.java
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.utils;
+
+import org.apache.fluss.utils.function.ThrowingSupplier;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.function.Predicate;
+
+/**
+ * Utility class for executing operations with retry logic.
+ *
+ * <p>This class provides methods to execute operations that may fail transiently, with configurable
+ * retry count and exponential backoff.
+ */
+public final class RetryUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RetryUtils.class);
+
+ private RetryUtils() {}
+
+ /**
+ * Executes an operation with retry logic using exponential backoff.
+ *
+ * @param operation the operation to execute
+ * @param operationName a descriptive name for logging
+ * @param maxRetries maximum number of retry attempts
+ * @param initialBackoffMs initial backoff delay in milliseconds
+ * @param maxBackoffMs maximum backoff delay in milliseconds
+ * @param retryPredicate predicate to determine if an exception should trigger a retry
+ * @param <T> the return type of the operation
+ * @return the result of the operation
+ * @throws IOException if all retries are exhausted or a non-retryable exception occurs
+ * @throws InterruptedIOException if the thread is interrupted during retry backoff
+ */
+ public static <T> T executeWithRetry(
+ ThrowingSupplier<T, ? extends Exception> operation,
+ String operationName,
+ int maxRetries,
+ long initialBackoffMs,
+ long maxBackoffMs,
+ Predicate<Exception> retryPredicate)
+ throws IOException {
+
+ Exception lastException = null;
+ long backoffMs = initialBackoffMs;
+
+ for (int attempt = 1; attempt <= maxRetries; attempt++) {
+ try {
+ return operation.get();
+ } catch (Exception e) {
+ // Check if the exception is an InterruptedException or wraps one
+ if (isInterruptedException(e)) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException(
+ operationName + " was interrupted on attempt " + attempt);
+ }
+
+ lastException = e;
+
+ if (!retryPredicate.test(e)) {
+ // Non-retryable exception
+ throw wrapAsIOException(e, operationName);
+ }
+
+ if (attempt < maxRetries) {
+ LOG.warn(
+ "{} failed (attempt {}/{}), retrying in {} ms",
+ operationName,
+ attempt,
+ maxRetries,
+ backoffMs,
+ e);
+ try {
+ Thread.sleep(backoffMs);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException(
+ operationName
+ + " was interrupted while waiting for retry after attempt "
+ + attempt);
+ }
+ backoffMs = Math.min(backoffMs * 2, maxBackoffMs);
+ }
+ }
+ }
+
+ throw new IOException(
+ String.format("%s failed after %d attempts", operationName, maxRetries),
+ lastException);
+ }
+
+ /**
+ * Executes an IO operation with retry logic. Retries on any IOException.
+ *
+ * @param operation the operation to execute
+ * @param operationName a descriptive name for logging
+ * @param maxRetries maximum number of retry attempts
+ * @param initialBackoffMs initial backoff delay in milliseconds
+ * @param maxBackoffMs maximum backoff delay in milliseconds
+ * @param <T> the return type of the operation
+ * @return the result of the operation
+ * @throws IOException if all retries are exhausted or a non-retryable exception occurs
+ * @throws InterruptedIOException if the thread is interrupted during retry backoff
+ */
+ public static <T> T executeIOWithRetry(
+ ThrowingSupplier<T, ? extends Exception> operation,
+ String operationName,
+ int maxRetries,
+ long initialBackoffMs,
+ long maxBackoffMs)
+ throws IOException {
+ return executeWithRetry(
+ operation,
+ operationName,
+ maxRetries,
+ initialBackoffMs,
+ maxBackoffMs,
+ e -> e instanceof IOException);
+ }
+
+ /**
+ * Checks if the exception is an InterruptedException or wraps one.
+ *
+ * @param e the exception to check
+ * @return true if the exception is or wraps an InterruptedException
+ */
+ private static boolean isInterruptedException(Exception e) {
+ Throwable current = e;
+ while (current != null) {
+ if (current instanceof InterruptedException) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+
+ private static IOException wrapAsIOException(Exception e, String operationName) {
+ if (e instanceof IOException) {
+ return (IOException) e;
+ }
+ return new IOException(operationName + " failed", e);
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/function/ThrowingSupplier.java b/fluss-common/src/main/java/org/apache/fluss/utils/function/ThrowingSupplier.java
new file mode 100644
index 000000000..e1dacc584
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/function/ThrowingSupplier.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.utils.function;
+
+/**
+ * A functional interface that supplies a result and may throw an exception.
+ *
+ * @param <T> the type of results supplied by this supplier
+ * @param <E> the type of exception that may be thrown
+ */
+@FunctionalInterface
+public interface ThrowingSupplier<T, E extends Exception> {
+
+ /**
+ * Gets a result.
+ *
+ * @return a result
+ * @throws E if unable to supply a result
+ */
+ T get() throws E;
+}
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/RetryUtilsTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/RetryUtilsTest.java
new file mode 100644
index 000000000..ae52b9d8b
--- /dev/null
+++ b/fluss-common/src/test/java/org/apache/fluss/utils/RetryUtilsTest.java
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.utils;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit tests for {@link RetryUtils}. */
+class RetryUtilsTest {
+
+ @Test
+ void testSuccessOnFirstAttempt() throws IOException {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ String result =
+ RetryUtils.executeWithRetry(
+ () -> {
+ attempts.incrementAndGet();
+ return "success";
+ },
+ "testOp",
+ 3,
+ 10,
+ 100,
+ e -> true);
+
+ assertThat(result).isEqualTo("success");
+ assertThat(attempts.get()).isEqualTo(1);
+ }
+
+ @Test
+ void testSuccessAfterRetries() throws IOException {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ String result =
+ RetryUtils.executeWithRetry(
+ () -> {
+ int attempt = attempts.incrementAndGet();
+ if (attempt < 3) {
+ throw new IOException("transient failure");
+ }
+ return "success";
+ },
+ "testOp",
+ 5,
+ 10,
+ 100,
+ e -> e instanceof IOException);
+
+ assertThat(result).isEqualTo("success");
+ assertThat(attempts.get()).isEqualTo(3);
+ }
+
+ @Test
+ void testExhaustedRetries() {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ assertThatThrownBy(
+ () ->
+ RetryUtils.executeWithRetry(
+ () -> {
+ attempts.incrementAndGet();
+ throw new IOException("persistent failure");
+ },
+ "testOp",
+ 3,
+ 10,
+ 100,
+ e -> e instanceof IOException))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("testOp failed after 3 attempts")
+ .hasCauseInstanceOf(IOException.class);
+
+ assertThat(attempts.get()).isEqualTo(3);
+ }
+
+ @Test
+ void testNonRetryableException() {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ assertThatThrownBy(
+ () ->
+ RetryUtils.executeWithRetry(
+ () -> {
+ attempts.incrementAndGet();
+ throw new IllegalArgumentException("bad argument");
+ },
+ "testOp",
+ 5,
+ 10,
+ 100,
+ e -> e instanceof IOException))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("testOp failed")
+ .hasCauseInstanceOf(IllegalArgumentException.class);
+
+ // Should fail immediately without retries
+ assertThat(attempts.get()).isEqualTo(1);
+ }
+
+ @Test
+ void testIOExceptionPassedThrough() {
+ IOException originalException = new IOException("original");
+
+ assertThatThrownBy(
+ () ->
+ RetryUtils.executeWithRetry(
+ () -> {
+ throw originalException;
+ },
+ "testOp",
+ 1,
+ 10,
+ 100,
+ e -> false))
+ .isInstanceOf(IOException.class)
+ .isSameAs(originalException);
+ }
+
+ @Test
+ void testExecuteIOWithRetry() throws IOException {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ String result =
+ RetryUtils.executeIOWithRetry(
+ () -> {
+ int attempt = attempts.incrementAndGet();
+ if (attempt < 2) {
+ throw new IOException("transient");
+ }
+ return "done";
+ },
+ "ioOp",
+ 3,
+ 10,
+ 100);
+
+ assertThat(result).isEqualTo("done");
+ assertThat(attempts.get()).isEqualTo(2);
+ }
+
+ @Test
+ void testExecuteIOWithRetryNonIOException() {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ assertThatThrownBy(
+ () ->
+ RetryUtils.executeIOWithRetry(
+ () -> {
+ attempts.incrementAndGet();
+ throw new RuntimeException("not IO");
+ },
+ "ioOp",
+ 3,
+ 10,
+ 100))
+ .isInstanceOf(IOException.class)
+ .hasCauseInstanceOf(RuntimeException.class);
+
+ // RuntimeException is not retryable for executeIOWithRetry
+ assertThat(attempts.get()).isEqualTo(1);
+ }
+
+ @Test
+ void testBackoffCapped() throws IOException {
+ AtomicInteger attempts = new AtomicInteger(0);
+ long startTime = System.currentTimeMillis();
+
+ RetryUtils.executeWithRetry(
+ () -> {
+ int attempt = attempts.incrementAndGet();
+ if (attempt < 4) {
+ throw new IOException("fail");
+ }
+ return "ok";
+ },
+ "backoffTest",
+ 5,
+ 10, // initial backoff
+ 50, // max backoff (caps exponential growth)
+ e -> e instanceof IOException);
+
+ long elapsed = System.currentTimeMillis() - startTime;
+ // 3 retries: 10ms + 20ms + 40ms (capped to 50ms) = 80ms minimum
+ // But with cap: 10ms + 20ms + 50ms = 80ms minimum
+ // Allow some tolerance for test execution
+ assertThat(elapsed).isGreaterThanOrEqualTo(70);
+ assertThat(attempts.get()).isEqualTo(4);
+ }
+
+ @Test
+ void testInterruptedExceptionFromOperation() {
+ AtomicInteger attempts = new AtomicInteger(0);
+
+ assertThatThrownBy(
+ () ->
+ RetryUtils.executeWithRetry(
+ () -> {
+ attempts.incrementAndGet();
+ throw new InterruptedException("interrupted");
+ },
+ "interruptOp",
+ 5,
+ 10,
+ 100,
+ e -> true))
+ .isInstanceOf(InterruptedIOException.class)
+ .hasMessageContaining("interruptOp was interrupted on attempt 1");
+
+ // Should fail immediately without retries
+ assertThat(attempts.get()).isEqualTo(1);
+ // Verify interrupt status is preserved
+ assertThat(Thread.currentThread().isInterrupted()).isTrue();
+ // Clear interrupt status for other tests
+ Thread.interrupted();
+ }
+}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java
index 072a5954f..3885bb387 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminGateway.java
@@ -33,6 +33,8 @@ import org.apache.fluss.rpc.messages.CreatePartitionRequest;
import org.apache.fluss.rpc.messages.CreatePartitionResponse;
import org.apache.fluss.rpc.messages.CreateTableRequest;
import org.apache.fluss.rpc.messages.CreateTableResponse;
+import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.DeleteProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.DropAclsRequest;
import org.apache.fluss.rpc.messages.DropAclsResponse;
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
@@ -41,10 +43,14 @@ import org.apache.fluss.rpc.messages.DropPartitionRequest;
import org.apache.fluss.rpc.messages.DropPartitionResponse;
import org.apache.fluss.rpc.messages.DropTableRequest;
import org.apache.fluss.rpc.messages.DropTableResponse;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.ListRebalanceProgressRequest;
import org.apache.fluss.rpc.messages.ListRebalanceProgressResponse;
import org.apache.fluss.rpc.messages.RebalanceRequest;
import org.apache.fluss.rpc.messages.RebalanceResponse;
+import org.apache.fluss.rpc.messages.RegisterProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.RegisterProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.RemoveServerTagRequest;
import org.apache.fluss.rpc.messages.RemoveServerTagResponse;
import org.apache.fluss.rpc.protocol.ApiKeys;
@@ -146,6 +152,40 @@ public interface AdminGateway extends AdminReadOnlyGateway {
@RPC(api = ApiKeys.CANCEL_REBALANCE)
CompletableFuture<CancelRebalanceResponse> cancelRebalance(CancelRebalanceRequest request);
+ // ==================================================================================
+ // Producer Offset Management APIs (for Exactly-Once Semantics)
+ // ==================================================================================
+
+ /**
+ * Register producer offset snapshot with atomic "check and register" semantics.
+ *
+ * @param request the request containing producer ID and offsets
+ * @return response indicating whether snapshot was created or already existed
+ */
+ @RPC(api = ApiKeys.REGISTER_PRODUCER_OFFSETS)
+ CompletableFuture<RegisterProducerOffsetsResponse> registerProducerOffsets(
+ RegisterProducerOffsetsRequest request);
+
+ /**
+ * Get producer offset snapshot.
+ *
+ * @param request the request containing producer ID
+ * @return response containing the producer offsets
+ */
+ @RPC(api = ApiKeys.GET_PRODUCER_OFFSETS)
+ CompletableFuture<GetProducerOffsetsResponse> getProducerOffsets(
+ GetProducerOffsetsRequest request);
+
+ /**
+ * Delete producer offset snapshot.
+ *
+ * @param request the request containing producer ID
+ * @return response indicating deletion success
+ */
+ @RPC(api = ApiKeys.DELETE_PRODUCER_OFFSETS)
+ CompletableFuture<DeleteProducerOffsetsResponse> deleteProducerOffsets(
+ DeleteProducerOffsetsRequest request);
+
// todo: rename table & alter table
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
index cc033ba8a..f962d346c 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
@@ -80,7 +80,10 @@ public enum ApiKeys {
REBALANCE(1049, 0, 0, PUBLIC),
LIST_REBALANCE_PROGRESS(1050, 0, 0, PUBLIC),
CANCEL_REBALANCE(1051, 0, 0, PUBLIC),
- PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 0, PRIVATE);
+ PREPARE_LAKE_TABLE_SNAPSHOT(1052, 0, 0, PRIVATE),
+ REGISTER_PRODUCER_OFFSETS(1053, 0, 0, PUBLIC),
+ GET_PRODUCER_OFFSETS(1054, 0, 0, PUBLIC),
+ DELETE_PRODUCER_OFFSETS(1055, 0, 0, PUBLIC);
private static final Map<Integer, ApiKeys> ID_TO_TYPE =
Arrays.stream(ApiKeys.values())
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
index 5ee652cce..36ecd819f 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/Errors.java
@@ -36,6 +36,7 @@ import org.apache.fluss.exception.InvalidConfigException;
import org.apache.fluss.exception.InvalidCoordinatorException;
import org.apache.fluss.exception.InvalidDatabaseException;
import org.apache.fluss.exception.InvalidPartitionException;
+import org.apache.fluss.exception.InvalidProducerIdException;
import org.apache.fluss.exception.InvalidReplicationFactorException;
import org.apache.fluss.exception.InvalidRequiredAcksException;
import org.apache.fluss.exception.InvalidServerRackInfoException;
@@ -240,7 +241,11 @@ public enum Errors {
SEVER_TAG_NOT_EXIST_EXCEPTION(60, "The server tag not exist.", ServerTagNotExistException::new),
REBALANCE_FAILURE_EXCEPTION(61, "The rebalance task failure.", RebalanceFailureException::new),
NO_REBALANCE_IN_PROGRESS_EXCEPTION(
- 62, "No rebalance task in progress.", NoRebalanceInProgressException::new);
+ 62, "No rebalance task in progress.", NoRebalanceInProgressException::new),
+ INVALID_PRODUCER_ID_EXCEPTION(
+ 63,
+ "The client has attempted to perform an operation with an invalid producer ID.",
+ InvalidProducerIdException::new);
private static final Logger LOG = LoggerFactory.getLogger(Errors.class);
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index a9cba3f47..d3b1edb96 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -623,6 +623,41 @@ message CancelRebalanceRequest {
message CancelRebalanceResponse {
}
+// ------------------------------------------------------------------------------------------
+// Producer Offset Management
+// ------------------------------------------------------------------------------------------
+
+// Register producer offsets request and response
+message RegisterProducerOffsetsRequest {
+ required string producer_id = 1;
+ repeated PbProducerTableOffsets table_offsets = 2;
+ optional int64 ttl_ms = 3;
+}
+
+message RegisterProducerOffsetsResponse {
+ // Result of registration: 0 = CREATED (new snapshot), 1 = ALREADY_EXISTS (snapshot existed)
+ optional int32 result = 1;
+}
+
+// Get producer offsets request and response
+message GetProducerOffsetsRequest {
+ required string producer_id = 1;
+}
+
+message GetProducerOffsetsResponse {
+ optional string producer_id = 1;
+ optional int64 expiration_time = 2;
+ repeated PbProducerTableOffsets table_offsets = 3;
+}
+
+// Delete producer offsets request and response
+message DeleteProducerOffsetsRequest {
+ required string producer_id = 1;
+}
+
+message DeleteProducerOffsetsResponse {
+}
+
// --------------- Inner classes ----------------
@@ -1068,4 +1103,10 @@ message PbBucketOffset {
optional int64 partition_id = 1;
required int32 bucket_id = 2;
optional int64 log_end_offset = 4;
+}
+
+// Helper message for table offsets in producer offsets (without table_path)
+message PbProducerTableOffsets {
+ required int64 table_id = 1;
+ repeated PbBucketOffset bucket_offsets = 2;
}
\ No newline at end of file
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
index b23fbb7ac..6a846076a 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java
@@ -26,6 +26,7 @@ import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.config.cluster.AlterConfig;
import org.apache.fluss.config.cluster.AlterConfigOpType;
+import org.apache.fluss.exception.ApiException;
import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidCoordinatorException;
import org.apache.fluss.exception.InvalidDatabaseException;
@@ -45,6 +46,7 @@ import org.apache.fluss.metadata.DeleteBehavior;
import org.apache.fluss.metadata.MergeEngineType;
import org.apache.fluss.metadata.PartitionSpec;
import org.apache.fluss.metadata.ResolvedPartitionSpec;
+import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
@@ -75,6 +77,8 @@ import org.apache.fluss.rpc.messages.CreatePartitionRequest;
import org.apache.fluss.rpc.messages.CreatePartitionResponse;
import org.apache.fluss.rpc.messages.CreateTableRequest;
import org.apache.fluss.rpc.messages.CreateTableResponse;
+import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.DeleteProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.DropAclsRequest;
import org.apache.fluss.rpc.messages.DropAclsResponse;
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
@@ -83,6 +87,8 @@ import org.apache.fluss.rpc.messages.DropPartitionRequest;
import org.apache.fluss.rpc.messages.DropPartitionResponse;
import org.apache.fluss.rpc.messages.DropTableRequest;
import org.apache.fluss.rpc.messages.DropTableResponse;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.LakeTieringHeartbeatRequest;
import org.apache.fluss.rpc.messages.LakeTieringHeartbeatResponse;
import org.apache.fluss.rpc.messages.ListRebalanceProgressRequest;
@@ -93,11 +99,14 @@ import org.apache.fluss.rpc.messages.PbAlterConfig;
import org.apache.fluss.rpc.messages.PbHeartbeatReqForTable;
import org.apache.fluss.rpc.messages.PbHeartbeatRespForTable;
import org.apache.fluss.rpc.messages.PbPrepareLakeTableRespForTable;
+import org.apache.fluss.rpc.messages.PbProducerTableOffsets;
import org.apache.fluss.rpc.messages.PbTableOffsets;
import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotRequest;
import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotResponse;
import org.apache.fluss.rpc.messages.RebalanceRequest;
import org.apache.fluss.rpc.messages.RebalanceResponse;
+import org.apache.fluss.rpc.messages.RegisterProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.RegisterProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.RemoveServerTagRequest;
import org.apache.fluss.rpc.messages.RemoveServerTagResponse;
import org.apache.fluss.rpc.netty.server.Session;
@@ -125,6 +134,7 @@ import org.apache.fluss.server.coordinator.event.EventManager;
import org.apache.fluss.server.coordinator.event.ListRebalanceProgressEvent;
import org.apache.fluss.server.coordinator.event.RebalanceEvent;
import org.apache.fluss.server.coordinator.event.RemoveServerTagEvent;
+import org.apache.fluss.server.coordinator.producer.ProducerOffsetsManager;
import org.apache.fluss.server.coordinator.rebalance.goal.Goal;
import org.apache.fluss.server.entity.CommitKvSnapshotData;
import org.apache.fluss.server.entity.LakeTieringTableInfo;
@@ -133,6 +143,7 @@ import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
import org.apache.fluss.server.kv.snapshot.CompletedSnapshotJsonSerde;
import org.apache.fluss.server.metadata.CoordinatorMetadataCache;
import org.apache.fluss.server.metadata.CoordinatorMetadataProvider;
+import org.apache.fluss.server.utils.ServerRpcMessageUtils;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.BucketAssignment;
import org.apache.fluss.server.zk.data.PartitionAssignment;
@@ -140,10 +151,14 @@ import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.server.zk.data.TableRegistration;
import org.apache.fluss.server.zk.data.lake.LakeTable;
import org.apache.fluss.server.zk.data.lake.LakeTableHelper;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsets;
import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.concurrent.FutureUtils;
import org.apache.fluss.utils.json.TableBucketOffsets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.annotation.Nullable;
import java.io.UncheckedIOException;
@@ -164,11 +179,13 @@ import static org.apache.fluss.config.FlussConfigUtils.isTableStorageConfig;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclBindingFilters;
import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclBindings;
import static org.apache.fluss.server.coordinator.rebalance.goal.GoalUtils.getGoalByType;
+import static org.apache.fluss.server.utils.ServerRpcMessageUtils.addTableOffsetsToResponse;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.fromTablePath;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getAdjustIsrData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getCommitLakeTableSnapshotData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getCommitRemoteLogManifestData;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.getPartitionSpec;
+import static org.apache.fluss.server.utils.ServerRpcMessageUtils.groupOffsetsByTableId;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeCreateAclsResponse;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.makeDropAclsResponse;
import static org.apache.fluss.server.utils.ServerRpcMessageUtils.toAlterTableConfigChanges;
@@ -181,6 +198,9 @@ import static org.apache.fluss.utils.Preconditions.checkNotNull;
/** An RPC Gateway service for coordinator server. */
public final class CoordinatorService extends RpcServiceBase implements CoordinatorGateway {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CoordinatorService.class);
+
private final int defaultBucketNumber;
private final int defaultReplicationFactor;
private final boolean logTableAllowCreation;
@@ -193,6 +213,7 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
private final LakeCatalogDynamicLoader lakeCatalogDynamicLoader;
private final ExecutorService ioExecutor;
private final LakeTableHelper lakeTableHelper;
+ private final ProducerOffsetsManager producerOffsetsManager;
public CoordinatorService(
Configuration conf,
@@ -228,6 +249,10 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
this.ioExecutor = ioExecutor;
this.lakeTableHelper =
new LakeTableHelper(zkClient, conf.getString(ConfigOptions.REMOTE_DATA_DIR));
+
+ // Initialize and start the producer snapshot manager
+ this.producerOffsetsManager = new ProducerOffsetsManager(conf, zkClient);
+ this.producerOffsetsManager.start();
}
@Override
@@ -237,32 +262,47 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
@Override
public void shutdown() {
+ IOUtils.closeQuietly(producerOffsetsManager, "producer snapshot manager");
IOUtils.closeQuietly(lakeCatalogDynamicLoader, "lake catalog");
}
@Override
public void authorizeTable(OperationType operationType, long tableId) {
if (authorizer != null) {
- TablePath tablePath;
- try {
- AccessContextEvent<TablePath> getTablePathEvent =
- new AccessContextEvent<>(ctx -> ctx.getTablePathById(tableId));
- eventManagerSupplier.get().put(getTablePathEvent);
- tablePath = getTablePathEvent.getResultFuture().get();
- } catch (Exception e) {
- throw new UnknownServerException("Failed to get table path by ID " + tableId, e);
- }
+ authorizeTableWithSession(currentSession(), operationType, tableId);
+ }
+ }
- if (tablePath == null) {
- throw new UnknownTableOrBucketException(
- String.format(
- "This server %s does not know this table ID %s. This may happen when the table "
- + "metadata cache in the server is not updated yet.",
- name(), tableId));
- }
+ /**
+ * Authorize table access with an explicitly provided session.
+ *
+ * <p>This method is used for async operations where the session must be captured before
+ * entering the async block, since currentSession() relies on thread-local storage.
+ */
+ private void authorizeTableWithSession(
+ Session session, OperationType operationType, long tableId) {
+ TablePath tablePath;
+ try {
+ // TODO: this will block on the coordinator event thread, consider refactor
+ // CoordinatorMetadataCache to hold the mapping of table_id to table_path, and then
+ // we don't need this async request.
+ AccessContextEvent<TablePath> getTablePathEvent =
+ new AccessContextEvent<>(ctx -> ctx.getTablePathById(tableId));
+ eventManagerSupplier.get().put(getTablePathEvent);
+ tablePath = getTablePathEvent.getResultFuture().get();
+ } catch (Exception e) {
+ throw new UnknownServerException("Failed to get table path by ID " + tableId, e);
+ }
- authorizeTable(operationType, tablePath);
+ if (tablePath == null) {
+ throw new UnknownTableOrBucketException(
+ String.format(
+ "This server %s does not know this table ID %s. This may happen when the table "
+ + "metadata cache in the server is not updated yet.",
+ name(), tableId));
}
+
+ authorizer.authorize(session, operationType, Resource.table(tablePath));
}
@Override
@@ -988,4 +1028,153 @@ public final class CoordinatorService extends RpcServiceBase implements Coordina
return flussPrincipal;
}
}
+
+ // ==================================================================================
+ // Producer Offset Management APIs (for Exactly-Once Semantics)
+ // ==================================================================================
+
+ @Override
+ public CompletableFuture<RegisterProducerOffsetsResponse> registerProducerOffsets(
+ RegisterProducerOffsetsRequest request) {
+ // Authorization: require WRITE permission on all tables in the request
+ if (authorizer != null) {
+ for (PbProducerTableOffsets tableOffsets : request.getTableOffsetsList()) {
+ long tableId = tableOffsets.getTableId();
+ authorizeTable(OperationType.WRITE, tableId);
+ }
+ }
+
+ return CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ String producerId = request.getProducerId();
+ Map<TableBucket, Long> offsets =
+ ServerRpcMessageUtils.toTableBucketOffsets(
+ request.getTableOffsetsList());
+
+ // Use custom TTL if provided, otherwise use default (null means use
+ // manager's default)
+ Long ttlMs = request.hasTtlMs() ? request.getTtlMs() : null;
+
+ // Register with atomic "check and register" semantics
+ boolean created =
+ producerOffsetsManager.registerSnapshot(producerId, offsets, ttlMs);
+
+ RegisterProducerOffsetsResponse response =
+ new RegisterProducerOffsetsResponse();
+ // Aligns with RegisterResult enum in fluss-client:
+ // 0 = CREATED, 1 = ALREADY_EXISTS
+ response.setResult(created ? 0 : 1);
+ return response;
+ } catch (ApiException e) {
+ // Re-throw ApiExceptions as-is to preserve exception type for client
+ throw e;
+ } catch (Exception e) {
+ throw new UnknownServerException(
+ "Failed to register producer offsets for producer "
+ + request.getProducerId(),
+ e);
+ }
+ },
+ ioExecutor);
+ }
+
+ @Override
+ public CompletableFuture<GetProducerOffsetsResponse> getProducerOffsets(
+ GetProducerOffsetsRequest request) {
+ String producerId = request.getProducerId();
+ // Capture session before entering async block since currentSession() is thread-local
+ Session session = authorizer != null ? currentSession() : null;
+
+ return CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ Optional<ProducerOffsets> optSnapshot =
+ producerOffsetsManager.getOffsetsMetadata(producerId);
+ if (!optSnapshot.isPresent()) {
+ return new GetProducerOffsetsResponse();
+ }
+
+ ProducerOffsets snapshot = optSnapshot.get();
+ Map<TableBucket, Long> allOffsets =
+ producerOffsetsManager.readOffsets(producerId);
+ Map<Long, Map<TableBucket, Long>> offsetsByTable =
+ groupOffsetsByTableId(allOffsets);
+
+ // Authorization: filter tables by READ permission
+ if (authorizer != null) {
+ offsetsByTable
+ .keySet()
+ .removeIf(
+ tableId -> {
+ try {
+ authorizeTableWithSession(
+ session, OperationType.READ, tableId);
+ return false; // keep this table
+ } catch (Exception e) {
+ return true; // remove this table
+ }
+ });
+ }
+
+ GetProducerOffsetsResponse response = new GetProducerOffsetsResponse();
+ response.setProducerId(producerId);
+ response.setExpirationTime(snapshot.getExpirationTime());
+
+ for (Map.Entry<Long, Map<TableBucket, Long>> entry :
+ offsetsByTable.entrySet()) {
+ addTableOffsetsToResponse(response, entry.getKey(), entry.getValue());
+ }
+
+ return response;
+ } catch (ApiException e) {
+ // Re-throw ApiExceptions as-is to preserve exception type for client
+ throw e;
+ } catch (Exception e) {
+ throw new UnknownServerException(
+ "Failed to get producer offsets for producer " + producerId, e);
+ }
+ },
+ ioExecutor);
+ }
+
+ @Override
+ public CompletableFuture<DeleteProducerOffsetsResponse> deleteProducerOffsets(
+ DeleteProducerOffsetsRequest request) {
+ // Capture session before entering async block since currentSession() is thread-local
+ Session session = authorizer != null ? currentSession() : null;
+ return CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ String producerId = request.getProducerId();
+
+ // Authorization: require WRITE permission on all tables in the snapshot
+ if (authorizer != null) {
+ Map<TableBucket, Long> offsets =
+ producerOffsetsManager.readOffsets(producerId);
+ // Extract unique table IDs from the snapshot
+ Set<Long> tableIds =
+ offsets.keySet().stream()
+ .map(TableBucket::getTableId)
+ .collect(Collectors.toSet());
+ // Check WRITE permission for each table
+ for (Long tableId : tableIds) {
+ authorizeTableWithSession(session, OperationType.WRITE, tableId);
+ }
+ }
+
+ producerOffsetsManager.deleteSnapshot(producerId);
+ return new DeleteProducerOffsetsResponse();
+ } catch (ApiException e) {
+ // Re-throw ApiExceptions as-is to preserve exception type for client
+ throw e;
+ } catch (Exception e) {
+ throw new UnknownServerException(
+ "Failed to delete producer offsets for producer "
+ + request.getProducerId(),
+ e);
+ }
+ },
+ ioExecutor);
+ }
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManager.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManager.java
new file mode 100644
index 000000000..e40fcb24c
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManager.java
@@ -0,0 +1,611 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.coordinator.producer;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InvalidProducerIdException;
+import org.apache.fluss.fs.FileStatus;
+import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsets;
+import org.apache.fluss.utils.concurrent.ExecutorThreadFactory;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Manager for producer offset snapshots lifecycle.
+ *
+ * <p>This manager handles:
+ *
+ * <ul>
+ * <li>Registering new producer snapshots with atomic "check and register" semantics
+ * <li>Retrieving producer snapshot offsets
+ * <li>Deleting producer snapshots
+ * <li>Periodic cleanup of expired snapshots
+ * <li>Cleanup of orphan files in remote storage
+ * </ul>
+ *
+ * <p>The manager uses a background scheduler to periodically clean up expired snapshots and orphan
+ * files. The cleanup interval and snapshot TTL are configurable.
+ *
+ * @see ProducerOffsetsStore for low-level storage operations
+ */
+public class ProducerOffsetsManager implements AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ProducerOffsetsManager.class);
+
+ /** Maximum number of attempts for snapshot registration to avoid infinite loops. */
+ private static final int MAX_REGISTER_ATTEMPTS = 3;
+
+ private final ProducerOffsetsStore offsetsStore;
+ private final long defaultTtlMs;
+ private final long cleanupIntervalMs;
+ private final ScheduledExecutorService cleanupScheduler;
+
+ public ProducerOffsetsManager(Configuration conf, ZooKeeperClient zkClient) {
+ this(
+ new ProducerOffsetsStore(zkClient, conf.getString(ConfigOptions.REMOTE_DATA_DIR)),
+ conf.get(ConfigOptions.COORDINATOR_PRODUCER_OFFSETS_TTL).toMillis(),
+ conf.get(ConfigOptions.COORDINATOR_PRODUCER_OFFSETS_CLEANUP_INTERVAL).toMillis());
+ }
+
+ @VisibleForTesting
+ ProducerOffsetsManager(
+ ProducerOffsetsStore offsetsStore, long defaultTtlMs, long cleanupIntervalMs) {
+ this.offsetsStore = offsetsStore;
+ this.defaultTtlMs = defaultTtlMs;
+ this.cleanupIntervalMs = cleanupIntervalMs;
+ this.cleanupScheduler =
+ Executors.newSingleThreadScheduledExecutor(
+ new ExecutorThreadFactory("producer-snapshot-cleanup"));
+ }
+
+ /** Starts the background cleanup task. */
+ public void start() {
+ cleanupScheduler.scheduleAtFixedRate(
+ this::runCleanup, cleanupIntervalMs, cleanupIntervalMs, TimeUnit.MILLISECONDS);
+ LOG.info(
+ "Started producer snapshot manager with TTL={} ms, cleanup interval={} ms",
+ defaultTtlMs,
+ cleanupIntervalMs);
+ }
+
+ // ------------------------------------------------------------------------
+ // Public API
+ // ------------------------------------------------------------------------
+
+ /**
+ * Registers a new producer offset snapshot with atomic "check and register" semantics.
+ *
+ * <p>This method uses ZooKeeper's version mechanism to handle concurrent requests safely:
+ *
+ * <ul>
+ * <li>Atomicity: Uses ZK's atomic create for new snapshots
+ * <li>Idempotency: Concurrent requests with same producerId will have exactly one succeed
+ * with CREATED, others return ALREADY_EXISTS
+ * <li>Version-based cleanup: Uses ZK version to safely delete expired snapshots without race
+ * conditions
+ * </ul>
+ *
+ * @param producerId the producer ID (typically Flink job ID)
+ * @param offsets map of TableBucket to offset for all tables
+ * @param ttlMs TTL in milliseconds for the snapshot, or null to use default
+ * @return true if a new snapshot was created (CREATED), false if snapshot already existed
+ * (ALREADY_EXISTS)
+ * @throws InvalidProducerIdException if producerId is invalid (null, empty, contains invalid
+ * characters, etc.)
+ * @throws Exception if the operation fails
+ */
+ public boolean registerSnapshot(String producerId, Map<TableBucket, Long> offsets, Long ttlMs)
+ throws Exception {
+ // Validate producerId as it will be used as both file name and ZK node name
+ validateProducerId(producerId);
+
+ long effectiveTtlMs = ttlMs != null ? ttlMs : defaultTtlMs;
+
+ // Use loop instead of recursion to avoid stack overflow risk
+ for (int attempt = 0; attempt < MAX_REGISTER_ATTEMPTS; attempt++) {
+ long currentTimeMs = System.currentTimeMillis();
+ long expirationTime = currentTimeMs + effectiveTtlMs;
+
+ // Step 1: Try to atomically create the snapshot (common case)
+ if (offsetsStore.tryStoreOffsets(producerId, offsets, expirationTime)) {
+ return true;
+ }
+
+ // Step 2: Snapshot exists - check if it's valid or expired
+ RegisterAttemptResult result =
+ handleExistingSnapshot(producerId, offsets, expirationTime, currentTimeMs);
+
+ switch (result) {
+ case ALREADY_EXISTS:
+ return false;
+ case CREATED:
+ return true;
+ case RETRY:
+ // Continue to next iteration
+ LOG.debug(
+ "Retrying snapshot registration for producer {} (attempt {}/{})",
+ producerId,
+ attempt + 1,
+ MAX_REGISTER_ATTEMPTS);
+ break;
+ }
+ }
+
+ // Exhausted all attempts
+ LOG.warn(
+ "Failed to register snapshot for producer {} after {} attempts, "
+ + "concurrent modifications may have interfered",
+ producerId,
+ MAX_REGISTER_ATTEMPTS);
+ return false;
+ }
+
+ /**
+ * Handles the case where a snapshot already exists for the producer.
+ *
+ * @return the result indicating next action: ALREADY_EXISTS, CREATED, or RETRY
+ */
+ private RegisterAttemptResult handleExistingSnapshot(
+ String producerId,
+ Map<TableBucket, Long> offsets,
+ long expirationTime,
+ long currentTimeMs)
+ throws Exception {
+
+ Optional<Tuple2<ProducerOffsets, Integer>> existingWithVersion =
+ offsetsStore.getOffsetsMetadataWithVersion(producerId);
+
+ // Case 1: Snapshot was deleted between our create attempt and this check
+ if (!existingWithVersion.isPresent()) {
+ return offsetsStore.tryStoreOffsets(producerId, offsets, expirationTime)
+ ? RegisterAttemptResult.CREATED
+ : RegisterAttemptResult.RETRY;
+ }
+
+ ProducerOffsets existingSnapshot = existingWithVersion.get().f0;
+ int version = existingWithVersion.get().f1;
+
+ // Case 2: Valid (non-expired) snapshot exists
+ if (!existingSnapshot.isExpired(currentTimeMs)) {
+ LOG.info(
+ "Producer snapshot already exists for producer {} (expires at {}, version {})",
+ producerId,
+ existingSnapshot.getExpirationTime(),
+ version);
+ return RegisterAttemptResult.ALREADY_EXISTS;
+ }
+
+ // Case 3: Expired snapshot - try to clean up and create new one
+ return tryReplaceExpiredSnapshot(
+ producerId, offsets, expirationTime, existingSnapshot, version, currentTimeMs);
+ }
+
+ /**
+ * Attempts to replace an expired snapshot with a new one.
+ *
+ * @return the result indicating next action
+ */
+ private RegisterAttemptResult tryReplaceExpiredSnapshot(
+ String producerId,
+ Map<TableBucket, Long> offsets,
+ long expirationTime,
+ ProducerOffsets expiredSnapshot,
+ int version,
+ long currentTimeMs)
+ throws Exception {
+
+ LOG.info(
+ "Found expired snapshot for producer {} (version {}), attempting cleanup",
+ producerId,
+ version);
+
+ // Try version-based conditional delete
+ boolean deleted =
+ offsetsStore.deleteSnapshotIfVersion(producerId, expiredSnapshot, version);
+
+ if (!deleted) {
+ // Version mismatch - another process modified the snapshot
+ LOG.debug(
+ "Version mismatch during delete for producer {}, checking current state",
+ producerId);
+ return checkCurrentSnapshotState(producerId, currentTimeMs);
+ }
+
+ // Successfully deleted, try to create new snapshot
+ if (offsetsStore.tryStoreOffsets(producerId, offsets, expirationTime)) {
+ return RegisterAttemptResult.CREATED;
+ }
+
+ // Another concurrent request created a snapshot after our delete
+ LOG.debug(
+ "Concurrent creation detected for producer {} after delete, checking state",
+ producerId);
+ return checkCurrentSnapshotState(producerId, currentTimeMs);
+ }
+
+ /**
+ * Checks the current snapshot state and returns the appropriate result.
+ *
+ * <p>This method is used after detecting concurrent modifications to determine whether a valid
+ * snapshot now exists or if we should retry.
+ *
+ * @param producerId the producer ID
+ * @param currentTimeMs the current time for expiration check
+ * @return ALREADY_EXISTS if a valid snapshot exists, RETRY otherwise
+ */
+ private RegisterAttemptResult checkCurrentSnapshotState(String producerId, long currentTimeMs)
+ throws Exception {
+ Optional<ProducerOffsets> producerOffsets = offsetsStore.getOffsetsMetadata(producerId);
+
+ if (isValidSnapshot(producerOffsets, currentTimeMs)) {
+ LOG.info(
+ "Valid snapshot exists for producer {} after concurrent modification",
+ producerId);
+ return RegisterAttemptResult.ALREADY_EXISTS;
+ }
+
+ LOG.debug("No valid snapshot for producer {} after concurrent modification", producerId);
+ return RegisterAttemptResult.RETRY;
+ }
+
+ /**
+ * Checks if a snapshot is present and not expired.
+ *
+ * @param producerOffsets the optional snapshot
+ * @param currentTimeMs the current time for expiration check
+ * @return true if snapshot exists and is not expired
+ */
+ private boolean isValidSnapshot(Optional<ProducerOffsets> producerOffsets, long currentTimeMs) {
+ return producerOffsets.isPresent() && !producerOffsets.get().isExpired(currentTimeMs);
+ }
+
+ /** Result of a single registration attempt. */
+ private enum RegisterAttemptResult {
+ /** Snapshot was successfully created. */
+ CREATED,
+ /** A valid snapshot already exists. */
+ ALREADY_EXISTS,
+ /** Should retry the registration. */
+ RETRY
+ }
+
+ /**
+ * Gets the offsets snapshot metadata for a producer.
+ *
+ * @param producerId the producer ID
+ * @return Optional containing the offsets snapshot if exists
+ * @throws InvalidProducerIdException if producerId is invalid
+ * @throws Exception if the operation fails
+ */
+ public Optional<ProducerOffsets> getOffsetsMetadata(String producerId) throws Exception {
+ validateProducerId(producerId);
+ return offsetsStore.getOffsetsMetadata(producerId);
+ }
+
+ /**
+ * Reads all offsets for a producer.
+ *
+ * @param producerId the producer ID
+ * @return map of TableBucket to offset, or empty map if no snapshot exists
+ * @throws InvalidProducerIdException if producerId is invalid
+ * @throws Exception if the operation fails
+ */
+ public Map<TableBucket, Long> readOffsets(String producerId) throws Exception {
+ validateProducerId(producerId);
+ return offsetsStore.readOffsets(producerId);
+ }
+
+ /**
+ * Deletes a producer snapshot.
+ *
+ * @param producerId the producer ID
+ * @throws InvalidProducerIdException if producerId is invalid
+ * @throws Exception if the operation fails
+ */
+ public void deleteSnapshot(String producerId) throws Exception {
+ validateProducerId(producerId);
+ offsetsStore.deleteSnapshot(producerId);
+ }
+
+ /**
+ * Gets the default TTL in milliseconds.
+ *
+ * @return the default TTL
+ */
+ public long getDefaultTtlMs() {
+ return defaultTtlMs;
+ }
+
+ // ------------------------------------------------------------------------
+ // Validation
+ // ------------------------------------------------------------------------
+
+ /**
+ * Validates that a producer ID is valid for use as both a file name and ZK node name.
+ *
+ * @param producerId the producer ID to validate
+ * @throws InvalidProducerIdException if producerId is invalid
+ */
+ private void validateProducerId(String producerId) {
+ String invalidReason = TablePath.detectInvalidName(producerId);
+ if (invalidReason != null) {
+ throw new InvalidProducerIdException(
+ "Invalid producer ID '" + producerId + "': " + invalidReason);
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Cleanup Operations
+ // ------------------------------------------------------------------------
+
+ /** Runs the cleanup task (expired snapshots and orphan files). */
+ @VisibleForTesting
+ void runCleanup() {
+ try {
+ int expiredCount = cleanupExpiredSnapshots();
+ if (expiredCount > 0) {
+ LOG.info("Producer snapshot cleanup: removed {} expired snapshots", expiredCount);
+ }
+
+ int orphanCount = cleanupOrphanFiles();
+ if (orphanCount > 0) {
+ LOG.info("Producer snapshot cleanup: removed {} orphan files", orphanCount);
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to cleanup producer snapshots", e);
+ }
+ }
+
+ /**
+ * Cleans up expired producer snapshots.
+ *
+ * @return number of snapshots cleaned up
+ * @throws Exception if the operation fails
+ */
+ @VisibleForTesting
+ int cleanupExpiredSnapshots() throws Exception {
+ List<String> producerIds = offsetsStore.listProducerIds();
+ int cleanedCount = 0;
+ long currentTime = System.currentTimeMillis();
+
+ for (String producerId : producerIds) {
+ try {
+ Optional<Tuple2<ProducerOffsets, Integer>> snapshotWithVersion =
+ offsetsStore.getOffsetsMetadataWithVersion(producerId);
+ if (!snapshotWithVersion.isPresent()) {
+ continue;
+ }
+
+ ProducerOffsets producerOffsets = snapshotWithVersion.get().f0;
+ int version = snapshotWithVersion.get().f1;
+
+ if (producerOffsets.isExpired(currentTime)) {
+ // Use version-based delete to avoid deleting snapshots updated just now
+ boolean deleted =
+ offsetsStore.deleteSnapshotIfVersion(
+ producerId, producerOffsets, version);
+ if (deleted) {
+ cleanedCount++;
+ LOG.debug(
+ "Cleaned up expired snapshot for producer {} (version {})",
+ producerId,
+ version);
+ } else {
+ LOG.debug(
+ "Skip cleanup for producer {} due to version mismatch (snapshot was updated)",
+ producerId);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to cleanup snapshot for producer {}", producerId, e);
+ }
+ }
+
+ return cleanedCount;
+ }
+
+ /**
+ * Cleans up orphan files in remote storage that don't have corresponding ZK metadata.
+ *
+ * <p>Orphan files can occur when:
+ *
+ * <ul>
+ * <li>ZK metadata was deleted but remote file deletion failed
+ * <li>Process crashed after writing remote file but before creating ZK metadata
+ * <li>Network issues caused partial cleanup
+ * </ul>
+ *
+ * @return number of orphan files cleaned up
+ * @throws Exception if the operation fails
+ */
+ @VisibleForTesting
+ int cleanupOrphanFiles() throws Exception {
+ FsPath producersDir = offsetsStore.getProducersDirectory();
+ FileSystem fileSystem = producersDir.getFileSystem();
+
+ if (!fileSystem.exists(producersDir)) {
+ LOG.debug("Producers directory does not exist, no orphan files to clean");
+ return 0;
+ }
+
+ // Get all producer IDs from ZK (these are valid)
+ Set<String> validProducerIds = new HashSet<>(offsetsStore.listProducerIds());
+
+ // Collect all valid file paths from ZK metadata
+ Set<String> validFilePaths = collectValidFilePaths(validProducerIds);
+
+ int cleanedCount = 0;
+
+ // List all producer directories in remote storage
+ FileStatus[] producerDirs = fileSystem.listStatus(producersDir);
+ if (producerDirs == null) {
+ return 0;
+ }
+
+ for (FileStatus producerDirStatus : producerDirs) {
+ if (!producerDirStatus.isDir()) {
+ continue;
+ }
+
+ String producerId = producerDirStatus.getPath().getName();
+
+ // Case 1: Producer ID not in ZK - entire directory is orphan
+ if (!validProducerIds.contains(producerId)) {
+ LOG.info(
+ "Found orphan producer directory {} (no ZK metadata), cleaning up",
+ producerId);
+ cleanedCount +=
+ offsetsStore.deleteDirectoryRecursively(
+ fileSystem, producerDirStatus.getPath());
+ continue;
+ }
+
+ // Case 2: Producer ID exists in ZK - check individual files
+ cleanedCount +=
+ cleanupOrphanFilesForProducer(
+ fileSystem, producerDirStatus.getPath(), validFilePaths);
+ }
+
+ return cleanedCount;
+ }
+
+ private Set<String> collectValidFilePaths(Set<String> validProducerIds) {
+ Set<String> validFilePaths = new HashSet<>();
+ for (String producerId : validProducerIds) {
+ try {
+ Optional<ProducerOffsets> optSnapshot = offsetsStore.getOffsetsMetadata(producerId);
+ if (optSnapshot.isPresent()) {
+ for (ProducerOffsets.TableOffsetMetadata tableMetadata :
+ optSnapshot.get().getTableOffsets()) {
+ validFilePaths.add(tableMetadata.getOffsetsPath().toString());
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to get snapshot for producer {} during orphan cleanup, "
+ + "skipping its files",
+ producerId,
+ e);
+ }
+ }
+ return validFilePaths;
+ }
+
+ private int cleanupOrphanFilesForProducer(
+ FileSystem fileSystem, FsPath producerDir, Set<String> validFilePaths)
+ throws IOException {
+ int cleanedCount = 0;
+
+ // List table directories under producer
+ FileStatus[] tableDirs = fileSystem.listStatus(producerDir);
+ if (tableDirs == null) {
+ return 0;
+ }
+
+ for (FileStatus tableDirStatus : tableDirs) {
+ if (!tableDirStatus.isDir()) {
+ continue;
+ }
+
+ // List offset files under table directory
+ FileStatus[] offsetFiles = fileSystem.listStatus(tableDirStatus.getPath());
+ if (offsetFiles == null) {
+ continue;
+ }
+
+ boolean hasValidFiles = false;
+ for (FileStatus fileStatus : offsetFiles) {
+ if (fileStatus.isDir()) {
+ continue;
+ }
+
+ String filePath = fileStatus.getPath().toString();
+ if (!validFilePaths.contains(filePath)) {
+ // This file is not referenced by any ZK metadata - it's orphan
+ LOG.debug("Deleting orphan file: {}", filePath);
+ try {
+ fileSystem.delete(fileStatus.getPath(), false);
+ cleanedCount++;
+ } catch (IOException e) {
+ LOG.warn("Failed to delete orphan file: {}", filePath, e);
+ }
+ } else {
+ hasValidFiles = true;
+ }
+ }
+
+ // If table directory is now empty, delete it
+ if (!hasValidFiles) {
+ tryDeleteEmptyDirectory(fileSystem, tableDirStatus.getPath());
+ }
+ }
+
+ // Check if producer directory is now empty
+ tryDeleteEmptyDirectory(fileSystem, producerDir);
+
+ return cleanedCount;
+ }
+
+ private void tryDeleteEmptyDirectory(FileSystem fileSystem, FsPath dir) {
+ try {
+ FileStatus[] remaining = fileSystem.listStatus(dir);
+ if (remaining == null || remaining.length == 0) {
+ fileSystem.delete(dir, false);
+ LOG.debug("Deleted empty directory: {}", dir);
+ }
+ } catch (IOException e) {
+ LOG.warn("Failed to delete empty directory: {}", dir, e);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (cleanupScheduler != null) {
+ cleanupScheduler.shutdown();
+ try {
+ if (!cleanupScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+ cleanupScheduler.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ cleanupScheduler.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+ LOG.info("Producer snapshot manager closed");
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsStore.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsStore.java
new file mode 100644
index 000000000..107ed40c8
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsStore.java
@@ -0,0 +1,390 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.coordinator.producer;
+
+import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.fs.FSDataInputStream;
+import org.apache.fluss.fs.FSDataOutputStream;
+import org.apache.fluss.fs.FileStatus;
+import org.apache.fluss.fs.FileSystem;
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsets;
+import org.apache.fluss.utils.FlussPaths;
+import org.apache.fluss.utils.IOUtils;
+import org.apache.fluss.utils.RetryUtils;
+import org.apache.fluss.utils.json.TableBucketOffsets;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
+/**
+ * Low-level storage operations for producer offset snapshots.
+ *
+ * <p>This class handles:
+ *
+ * <ul>
+ * <li>Writing offset data to remote storage (OSS/S3/HDFS)
+ * <li>Reading offset data from remote storage
+ * <li>Registering snapshot metadata in ZooKeeper
+ * <li>Deleting snapshots from both ZK and remote storage
+ * </ul>
+ *
+ * <p>This class is stateless and thread-safe. Lifecycle management should be handled by {@link
+ * ProducerOffsetsManager}.
+ */
+public class ProducerOffsetsStore {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ProducerOffsetsStore.class);
+
+ private static final int MAX_RETRIES = 3;
+ private static final long INITIAL_BACKOFF_MS = 100;
+ private static final long MAX_BACKOFF_MS = 1000;
+
+ private final ZooKeeperClient zkClient;
+ private final String remoteDataDir;
+
+ public ProducerOffsetsStore(ZooKeeperClient zkClient, String remoteDataDir) {
+ this.zkClient = zkClient;
+ this.remoteDataDir = remoteDataDir;
+ }
+
+ // ------------------------------------------------------------------------
+ // Core CRUD Operations
+ // ------------------------------------------------------------------------
+
+ /**
+ * Atomically stores a new producer offset snapshot.
+ *
+ * <p>This method first writes offset files to remote storage, then attempts to atomically
+ * create the ZK metadata. If the ZK node already exists, this method returns false and cleans
+ * up the remote files.
+ *
+ * <p><b>Note on potential orphan files:</b> There is a small window where orphan files can be
+ * created if the process crashes after writing remote files but before creating ZK metadata, or
+ * if ZK metadata creation fails. This is an acceptable trade-off because:
+ *
+ * <ul>
+ * <li>The alternative (ZK first, then files) would leave ZK metadata pointing to non-existent
+ * files, which is worse
+ * <li>Orphan files are harmless and will be cleaned up by {@link
+ * ProducerOffsetsManager#cleanupOrphanFiles()}
+ * <li>A unified orphan file cleanup mechanism will handle these cases in the future
+ * </ul>
+ *
+ * @param producerId the producer ID
+ * @param offsets map of TableBucket to offset
+ * @param expirationTime the expiration timestamp in milliseconds
+ * @return true if created, false if already existed
+ * @throws IOException if file operations fail
+ * @throws Exception if ZK operations fail
+ */
+ public boolean tryStoreOffsets(
+ String producerId, Map<TableBucket, Long> offsets, long expirationTime)
+ throws Exception {
+
+ Map<Long, Map<TableBucket, Long>> offsetsByTable = groupOffsetsByTable(offsets);
+ List<ProducerOffsets.TableOffsetMetadata> tableMetadatas = new ArrayList<>();
+ List<FsPath> createdFiles = new ArrayList<>();
+
+ try {
+ // Write offset files to remote storage
+ for (Map.Entry<Long, Map<TableBucket, Long>> entry : offsetsByTable.entrySet()) {
+ FsPath path = writeOffsetsFile(producerId, entry.getKey(), entry.getValue());
+ createdFiles.add(path);
+ tableMetadatas.add(new ProducerOffsets.TableOffsetMetadata(entry.getKey(), path));
+ }
+
+ // Atomically create ZK metadata
+ ProducerOffsets producerOffsets = new ProducerOffsets(expirationTime, tableMetadatas);
+ boolean created = zkClient.tryRegisterProducerOffsets(producerId, producerOffsets);
+
+ if (!created) {
+ LOG.info(
+ "Snapshot already exists for producer {}, cleaning up {} files",
+ producerId,
+ createdFiles.size());
+ cleanupFilesSafely(createdFiles);
+ return false;
+ }
+
+ LOG.info(
+ "Stored snapshot for producer {} with {} tables, expires at {}",
+ producerId,
+ tableMetadatas.size(),
+ expirationTime);
+ return true;
+
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to store snapshot for producer {}, cleaning up {} files",
+ producerId,
+ createdFiles.size(),
+ e);
+ cleanupFilesSafely(createdFiles);
+ throw e;
+ }
+ }
+
+ /** Gets the offset snapshot metadata for a producer. */
+ public Optional<ProducerOffsets> getOffsetsMetadata(String producerId) throws Exception {
+ return zkClient.getProducerOffsets(producerId);
+ }
+
+ /**
+ * Gets the offset snapshot metadata for a producer along with its ZK version.
+ *
+ * <p>The version can be used for conditional deletes to handle concurrent modifications safely.
+ *
+ * @param producerId the producer ID
+ * @return Optional containing a Tuple2 of (ProducerOffsets, version) if exists
+ * @throws Exception if the operation fails
+ */
+ public Optional<Tuple2<ProducerOffsets, Integer>> getOffsetsMetadataWithVersion(
+ String producerId) throws Exception {
+ return zkClient.getProducerOffsetsWithVersion(producerId);
+ }
+
+ /**
+ * Reads all offsets for a producer from remote storage.
+ *
+ * @throws IOException if file reading fails
+ * @throws Exception if ZK operations fail
+ */
+ public Map<TableBucket, Long> readOffsets(String producerId) throws Exception {
+ Optional<ProducerOffsets> optSnapshot = zkClient.getProducerOffsets(producerId);
+ if (!optSnapshot.isPresent()) {
+ return new HashMap<>();
+ }
+
+ Map<TableBucket, Long> allOffsets = new HashMap<>();
+ for (ProducerOffsets.TableOffsetMetadata metadata : optSnapshot.get().getTableOffsets()) {
+ allOffsets.putAll(readOffsetsFile(metadata.getOffsetsPath()));
+ }
+ return allOffsets;
+ }
+
+ /** Deletes a producer snapshot (both ZK metadata and remote files). */
+ public void deleteSnapshot(String producerId) throws Exception {
+ Optional<ProducerOffsets> optSnapshot = zkClient.getProducerOffsets(producerId);
+ if (!optSnapshot.isPresent()) {
+ LOG.debug("No snapshot found for producer {}", producerId);
+ return;
+ }
+
+ // Delete remote files
+ for (ProducerOffsets.TableOffsetMetadata metadata : optSnapshot.get().getTableOffsets()) {
+ deleteRemoteFile(metadata.getOffsetsPath());
+ }
+
+ // Delete ZK metadata
+ zkClient.deleteProducerOffsets(producerId);
+ LOG.info("Deleted snapshot for producer {}", producerId);
+ }
+
+ /**
+ * Deletes a producer snapshot only if the ZK version matches.
+ *
+ * <p>This provides optimistic concurrency control - the delete will only succeed if no other
+ * process has modified the snapshot since it was read.
+ *
+ * @param producerId the producer ID
+ * @param producerOffsets the snapshot to delete (used to get file paths)
+ * @param expectedVersion the expected ZK version
+ * @return true if deleted successfully, false if version mismatch
+ * @throws Exception if the operation fails for reasons other than version mismatch
+ */
+ public boolean deleteSnapshotIfVersion(
+ String producerId, ProducerOffsets producerOffsets, int expectedVersion)
+ throws Exception {
+ // First try to delete ZK metadata with version check
+ boolean deleted = zkClient.deleteProducerSnapshotIfVersion(producerId, expectedVersion);
+ if (!deleted) {
+ LOG.debug(
+ "Failed to delete snapshot for producer {} - version mismatch, "
+ + "snapshot was modified by another process",
+ producerId);
+ return false;
+ }
+
+ // ZK metadata deleted successfully, now clean up remote files
+ // Even if file deletion fails, the snapshot is already gone from ZK
+ // Orphan files will be cleaned up by the periodic cleanup task
+ for (ProducerOffsets.TableOffsetMetadata metadata : producerOffsets.getTableOffsets()) {
+ deleteRemoteFile(metadata.getOffsetsPath());
+ }
+
+ LOG.info("Deleted snapshot for producer {} with version {}", producerId, expectedVersion);
+ return true;
+ }
+
+ /** Lists all producer IDs with registered snapshots. */
+ public List<String> listProducerIds() throws Exception {
+ return zkClient.listProducerIds();
+ }
+
+ // ------------------------------------------------------------------------
+ // Remote Storage Operations
+ // ------------------------------------------------------------------------
+
+ /** Gets the base directory for producer snapshots in remote storage. */
+ public FsPath getProducersDirectory() {
+ return FlussPaths.remoteProducersDir(remoteDataDir);
+ }
+
+ /**
+ * Deletes a remote file, logging but not throwing on failure.
+ *
+ * @param filePath the file path to delete
+ * @return true if deleted successfully or file didn't exist, false if deletion failed
+ */
+ public boolean deleteRemoteFile(FsPath filePath) {
+ try {
+ FileSystem fs = filePath.getFileSystem();
+ if (fs.exists(filePath)) {
+ fs.delete(filePath, false);
+ LOG.debug("Deleted remote file: {}", filePath);
+ }
+ return true;
+ } catch (IOException e) {
+ LOG.warn("Failed to delete remote file: {}", filePath, e);
+ return false;
+ }
+ }
+
+ /** Recursively deletes a directory, returning the count of deleted files. */
+ public int deleteDirectoryRecursively(FileSystem fs, FsPath path) {
+ int count = 0;
+ try {
+ FileStatus[] contents = fs.listStatus(path);
+ if (contents != null) {
+ for (FileStatus status : contents) {
+ if (status.isDir()) {
+ count += deleteDirectoryRecursively(fs, status.getPath());
+ } else if (fs.delete(status.getPath(), false)) {
+ count++;
+ }
+ }
+ }
+ fs.delete(path, false);
+ } catch (IOException e) {
+ LOG.warn("Failed to delete directory: {}", path, e);
+ }
+ return count;
+ }
+
+ // ------------------------------------------------------------------------
+ // Private Helpers
+ // ------------------------------------------------------------------------
+
+ private Map<Long, Map<TableBucket, Long>> groupOffsetsByTable(Map<TableBucket, Long> offsets) {
+ Map<Long, Map<TableBucket, Long>> result = new HashMap<>();
+ for (Map.Entry<TableBucket, Long> entry : offsets.entrySet()) {
+ result.computeIfAbsent(entry.getKey().getTableId(), k -> new HashMap<>())
+ .put(entry.getKey(), entry.getValue());
+ }
+ return result;
+ }
+
+ private FsPath writeOffsetsFile(String producerId, long tableId, Map<TableBucket, Long> offsets)
+ throws IOException {
+
+ FsPath path =
+ FlussPaths.remoteProducerOffsetsPath(
+ remoteDataDir, producerId, tableId, UUID.randomUUID());
+
+ byte[] data = new TableBucketOffsets(tableId, offsets).toJsonBytes();
+
+ return RetryUtils.executeIOWithRetry(
+ () -> {
+ FileSystem fs = path.getFileSystem();
+ FsPath parentDir = path.getParent();
+ if (parentDir != null && !fs.exists(parentDir)) {
+ fs.mkdirs(parentDir);
+ }
+ try (FSDataOutputStream out = fs.create(path, FileSystem.WriteMode.OVERWRITE)) {
+ out.write(data);
+ }
+ LOG.debug(
+ "Wrote offsets for producer {} table {} at {}",
+ producerId,
+ tableId,
+ path);
+ return path;
+ },
+ String.format("write offsets file for producer %s table %d", producerId, tableId),
+ MAX_RETRIES,
+ INITIAL_BACKOFF_MS,
+ MAX_BACKOFF_MS);
+ }
+
+ private Map<TableBucket, Long> readOffsetsFile(FsPath path) throws IOException {
+ return RetryUtils.executeIOWithRetry(
+ () -> {
+ FileSystem fs = path.getFileSystem();
+ if (!fs.exists(path)) {
+ // intend to use FlussRuntimeException here to avoid retrying
+ throw new FlussRuntimeException("Offsets file not found: " + path);
+ }
+ try (FSDataInputStream in = fs.open(path);
+ ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+ IOUtils.copyBytes(in, out, true);
+ return TableBucketOffsets.fromJsonBytes(out.toByteArray()).getOffsets();
+ }
+ },
+ "read offsets file " + path,
+ MAX_RETRIES,
+ INITIAL_BACKOFF_MS,
+ MAX_BACKOFF_MS);
+ }
+
+ /**
+ * Safely cleans up a list of files, ensuring all files are attempted even if some fail.
+ *
+ * <p>This method reuses {@link #deleteRemoteFile(FsPath)} for each file and logs a summary if
+ * any deletions failed.
+ *
+ * @param files the list of files to clean up
+ */
+ private void cleanupFilesSafely(List<FsPath> files) {
+ int failedCount = 0;
+ for (FsPath file : files) {
+ if (!deleteRemoteFile(file)) {
+ failedCount++;
+ }
+ }
+ if (failedCount > 0) {
+ LOG.warn(
+ "Failed to cleanup {} out of {} files, orphan files will be cleaned by periodic task",
+ failedCount,
+ files.size());
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index f174f727a..a7bed51b8 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -68,6 +68,7 @@ import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse;
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataResponse;
import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsResponse;
import org.apache.fluss.rpc.messages.GetLatestLakeSnapshotResponse;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.InitWriterResponse;
import org.apache.fluss.rpc.messages.LakeTieringHeartbeatResponse;
import org.apache.fluss.rpc.messages.LimitScanResponse;
@@ -122,6 +123,7 @@ import org.apache.fluss.rpc.messages.PbPrefixLookupReqForBucket;
import org.apache.fluss.rpc.messages.PbPrefixLookupRespForBucket;
import org.apache.fluss.rpc.messages.PbProduceLogReqForBucket;
import org.apache.fluss.rpc.messages.PbProduceLogRespForBucket;
+import org.apache.fluss.rpc.messages.PbProducerTableOffsets;
import org.apache.fluss.rpc.messages.PbPutKvReqForBucket;
import org.apache.fluss.rpc.messages.PbPutKvRespForBucket;
import org.apache.fluss.rpc.messages.PbRebalancePlanForBucket;
@@ -1624,6 +1626,71 @@ public class ServerRpcMessageUtils {
return new TableBucketOffsets(tableId, bucketOffsets);
}
+ /**
+ * Populates a PbTableOffsets with bucket offsets.
+ *
+ * @param pbTableOffsets the protobuf table offsets to populate
+ * @param bucketOffsets map of TableBucket to offset
+ */
+ public static void populatePbTableOffsets(
+ PbTableOffsets pbTableOffsets, Map<TableBucket, Long> bucketOffsets) {
+ for (Map.Entry<TableBucket, Long> entry : bucketOffsets.entrySet()) {
+ TableBucket bucket = entry.getKey();
+ PbBucketOffset pbBucketOffset =
+ pbTableOffsets
+ .addBucketOffset()
+ .setBucketId(bucket.getBucket())
+ .setLogEndOffset(entry.getValue());
+ if (bucket.getPartitionId() != null) {
+ pbBucketOffset.setPartitionId(bucket.getPartitionId());
+ }
+ }
+ }
+
+ /**
+ * Groups offsets by table ID.
+ *
+ * @param allOffsets map of TableBucket to offset
+ * @return map of tableId to (map of TableBucket to offset)
+ */
+ public static Map<Long, Map<TableBucket, Long>> groupOffsetsByTableId(
+ Map<TableBucket, Long> allOffsets) {
+ Map<Long, Map<TableBucket, Long>> offsetsByTable = new HashMap<>();
+ for (Map.Entry<TableBucket, Long> entry : allOffsets.entrySet()) {
+ TableBucket bucket = entry.getKey();
+ offsetsByTable
+ .computeIfAbsent(bucket.getTableId(), k -> new HashMap<>())
+ .put(bucket, entry.getValue());
+ }
+ return offsetsByTable;
+ }
+
+ /**
+ * Populates a GetProducerOffsetsResponse with table offsets.
+ *
+ * @param response the response to populate
+ * @param tableId the table ID
+ * @param bucketOffsets the bucket offsets for this table
+ */
+ public static void addTableOffsetsToResponse(
+ GetProducerOffsetsResponse response,
+ long tableId,
+ Map<TableBucket, Long> bucketOffsets) {
+ PbProducerTableOffsets pbTableOffsets = response.addTableOffset();
+ pbTableOffsets.setTableId(tableId);
+ for (Map.Entry<TableBucket, Long> entry : bucketOffsets.entrySet()) {
+ TableBucket bucket = entry.getKey();
+ PbBucketOffset pbBucketOffset =
+ pbTableOffsets
+ .addBucketOffset()
+ .setBucketId(bucket.getBucket())
+ .setLogEndOffset(entry.getValue());
+ if (bucket.getPartitionId() != null) {
+ pbBucketOffset.setPartitionId(bucket.getPartitionId());
+ }
+ }
+ }
+
public static PbNotifyLakeTableOffsetReqForBucket makeNotifyLakeTableOffsetForBucket(
TableBucket tableBucket,
LakeTableSnapshot lakeTableSnapshot,
@@ -1882,4 +1949,26 @@ public class ServerRpcMessageUtils {
result.addAll(errors);
return result;
}
+
+ /**
+ * Converts a list of PbProducerTableOffsets to a map of TableBucket to offset.
+ *
+ * @param tableOffsetsList the list of PbProducerTableOffsets
+ * @return map of TableBucket to offset
+ */
+ public static Map<TableBucket, Long> toTableBucketOffsets(
+ List<PbProducerTableOffsets> tableOffsetsList) {
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ for (PbProducerTableOffsets pbTableOffsets : tableOffsetsList) {
+ long tableId = pbTableOffsets.getTableId();
+ for (PbBucketOffset pbBucketOffset : pbTableOffsets.getBucketOffsetsList()) {
+ Long partitionId =
+ pbBucketOffset.hasPartitionId() ? pbBucketOffset.getPartitionId() : null;
+ TableBucket bucket =
+ new TableBucket(tableId, partitionId, pbBucketOffset.getBucketId());
+ offsets.put(bucket, pbBucketOffset.getLogEndOffset());
+ }
+ }
+ return offsets;
+ }
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
index bd6bcad7f..4d9bf283d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java
@@ -66,6 +66,8 @@ import org.apache.fluss.server.zk.data.ZkData.PartitionIdZNode;
import org.apache.fluss.server.zk.data.ZkData.PartitionSequenceIdZNode;
import org.apache.fluss.server.zk.data.ZkData.PartitionZNode;
import org.apache.fluss.server.zk.data.ZkData.PartitionsZNode;
+import org.apache.fluss.server.zk.data.ZkData.ProducerIdZNode;
+import org.apache.fluss.server.zk.data.ZkData.ProducersZNode;
import org.apache.fluss.server.zk.data.ZkData.RebalanceZNode;
import org.apache.fluss.server.zk.data.ZkData.ResourceAclNode;
import org.apache.fluss.server.zk.data.ZkData.SchemaZNode;
@@ -80,6 +82,7 @@ import org.apache.fluss.server.zk.data.ZkData.TablesZNode;
import org.apache.fluss.server.zk.data.ZkData.WriterIdZNode;
import org.apache.fluss.server.zk.data.lake.LakeTable;
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsets;
import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework;
import org.apache.fluss.shaded.curator5.org.apache.curator.framework.api.BackgroundCallback;
import org.apache.fluss.shaded.curator5.org.apache.curator.framework.api.CuratorEvent;
@@ -1590,4 +1593,135 @@ public class ZooKeeperClient implements AutoCloseable {
}
return result;
}
+
+ // --------------------------------------------------------------------------------------------
+ // Producer Offset Snapshot
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Tries to atomically register a producer offset snapshot to ZK.
+ *
+ * <p>This method leverages ZooKeeper's atomic create operation to ensure that only one
+ * concurrent request can successfully create the snapshot. If a snapshot already exists for the
+ * given producer ID, this method returns false instead of throwing an exception.
+ *
+ * @param producerId the producer ID (typically Flink job ID)
+ * @param producerOffsets the producer offsets containing expiration time and table offset
+ * metadata
+ * @return true if the snapshot was created successfully, false if a snapshot already exists
+ * @throws Exception if the operation fails for reasons other than node already existing
+ */
+ public boolean tryRegisterProducerOffsets(String producerId, ProducerOffsets producerOffsets)
+ throws Exception {
+ String path = ProducerIdZNode.path(producerId);
+ try {
+ zkClient.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, ProducerIdZNode.encode(producerOffsets));
+ LOG.info("Registered producer snapshot for producer {} at path {}.", producerId, path);
+ return true;
+ } catch (KeeperException.NodeExistsException e) {
+ LOG.debug(
+ "Producer snapshot already exists for producer {} at path {}, "
+ + "returning false.",
+ producerId,
+ path);
+ return false;
+ }
+ }
+
+ /**
+ * Gets the {@link ProducerOffsets} for the given producer ID.
+ *
+ * @param producerId the producer ID
+ * @return an Optional containing the ProducerOffsets if it exists, empty otherwise
+ * @throws Exception if the operation fails
+ */
+ public Optional<ProducerOffsets> getProducerOffsets(String producerId) throws Exception {
+ String zkPath = ProducerIdZNode.path(producerId);
+ return getOrEmpty(zkPath).map(ProducerIdZNode::decode);
+ }
+
+ /**
+ * Deletes the producer offset snapshot for the given producer ID.
+ *
+ * @param producerId the producer ID
+ * @throws Exception if the operation fails
+ */
+ public void deleteProducerOffsets(String producerId) throws Exception {
+ String path = ProducerIdZNode.path(producerId);
+ zkClient.delete().forPath(path);
+ LOG.info("Deleted producer offsets snapshot for producer {} at path {}.", producerId, path);
+ }
+
+ /**
+ * Gets the {@link ProducerOffsets} for the given producer ID along with its ZK version.
+ *
+ * <p>The version can be used for conditional updates/deletes to handle concurrent modifications
+ * safely.
+ *
+ * @param producerId the producer ID
+ * @return an Optional containing a Tuple2 of (ProducerOffsets, version) if it exists, empty
+ * otherwise
+ * @throws Exception if the operation fails
+ */
+ public Optional<Tuple2<ProducerOffsets, Integer>> getProducerOffsetsWithVersion(
+ String producerId) throws Exception {
+ String zkPath = ProducerIdZNode.path(producerId);
+ try {
+ Stat stat = new Stat();
+ byte[] data = zkClient.getData().storingStatIn(stat).forPath(zkPath);
+ return Optional.of(Tuple2.of(ProducerIdZNode.decode(data), stat.getVersion()));
+ } catch (KeeperException.NoNodeException e) {
+ return Optional.empty();
+ }
+ }
+
+ /**
+ * Deletes the producer offset snapshot for the given producer ID only if the version matches.
+ *
+ * <p>This provides optimistic concurrency control - the delete will only succeed if no other
+ * process has modified the snapshot since it was read.
+ *
+ * @param producerId the producer ID
+ * @param expectedVersion the expected ZK version (obtained from getProducerSnapshotWithVersion)
+ * @return true if deleted successfully, false if version mismatch (snapshot was modified)
+ * @throws Exception if the operation fails for reasons other than version mismatch
+ */
+ public boolean deleteProducerSnapshotIfVersion(String producerId, int expectedVersion)
+ throws Exception {
+ String path = ProducerIdZNode.path(producerId);
+ try {
+ zkClient.delete().withVersion(expectedVersion).forPath(path);
+ LOG.info(
+ "Deleted producer snapshot for producer {} at path {} with version {}.",
+ producerId,
+ path,
+ expectedVersion);
+ return true;
+ } catch (KeeperException.BadVersionException e) {
+ LOG.debug(
+ "Failed to delete producer snapshot for producer {} - version mismatch "
+ + "(expected {}, snapshot was modified by another process).",
+ producerId,
+ expectedVersion);
+ return false;
+ } catch (KeeperException.NoNodeException e) {
+ LOG.debug(
+ "Producer snapshot for producer {} was already deleted by another process.",
+ producerId);
+ return true; // Already deleted, consider it success
+ }
+ }
+
+ /**
+ * Lists all producer IDs that have registered snapshots.
+ *
+ * @return list of producer IDs
+ * @throws Exception if the operation fails
+ */
+ public List<String> listProducerIds() throws Exception {
+ return getChildren(ProducersZNode.path());
+ }
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java
index 467245567..43c1acbf4 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/ZkData.java
@@ -26,6 +26,8 @@ import org.apache.fluss.security.acl.Resource;
import org.apache.fluss.security.acl.ResourceType;
import org.apache.fluss.server.zk.data.lake.LakeTable;
import org.apache.fluss.server.zk.data.lake.LakeTableJsonSerde;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsets;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsetsJsonSerde;
import org.apache.fluss.utils.json.JsonSerdeUtils;
import org.apache.fluss.utils.types.Tuple2;
@@ -840,4 +842,68 @@ public final class ZkData {
return JsonSerdeUtils.readValue(json, RebalanceTaskJsonSerde.INSTANCE);
}
}
+
+ // ------------------------------------------------------------------------------------------
+ // ZNodes under "/producers/"
+ // ------------------------------------------------------------------------------------------
+
+ /**
+ * The znode for producers. This is the root node for all producer offset snapshots. The znode
+ * path is:
+ *
+ * <p>/producers
+ */
+ public static final class ProducersZNode {
+ public static String path() {
+ return "/producers";
+ }
+ }
+
+ /**
+ * The znode for a specific producer's offset snapshot. The znode path is:
+ *
+ * <p>/producers/[producerId]
+ *
+ * <p>This znode stores {@link ProducerOffsets} which contains:
+ *
+ * <ul>
+ * <li>expiration_time: TTL for automatic cleanup
+ * <li>tables: List of table offset metadata with paths to remote offset files
+ * </ul>
+ *
+ * <p>The actual offset data is stored in remote storage (e.g., OSS/S3) and referenced by the
+ * file paths in the metadata.
+ */
+ public static final class ProducerIdZNode {
+ /**
+ * Returns the ZK path for the producer snapshot znode.
+ *
+ * @param producerId the producer ID (typically Flink job ID)
+ * @return the ZK path
+ */
+ public static String path(String producerId) {
+ return ProducersZNode.path() + "/" + producerId;
+ }
+
+ /**
+ * Encodes a ProducerOffsets to JSON bytes for storage in ZK.
+ *
+ * @param producerOffsets the ProducerOffsets to encode
+ * @return the encoded bytes
+ */
+ public static byte[] encode(ProducerOffsets producerOffsets) {
+ return JsonSerdeUtils.writeValueAsBytes(
+ producerOffsets, ProducerOffsetsJsonSerde.INSTANCE);
+ }
+
+ /**
+ * Decodes JSON bytes from ZK to a ProducerOffsets.
+ *
+ * @param json the JSON bytes from ZK
+ * @return the decoded ProducerOffsets
+ */
+ public static ProducerOffsets decode(byte[] json) {
+ return JsonSerdeUtils.readValue(json, ProducerOffsetsJsonSerde.INSTANCE);
+ }
+ }
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/producer/ProducerOffsets.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/producer/ProducerOffsets.java
new file mode 100644
index 000000000..bf8e2d225
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/producer/ProducerOffsets.java
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.zk.data.producer;
+
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.server.zk.data.ZkData;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Represents producer offset snapshot metadata stored in {@link ZkData.ProducerIdZNode}.
+ *
+ * <p>This class stores metadata about producer offset snapshots, including:
+ *
+ * <ul>
+ * <li>Expiration time for TTL management
+ * <li>List of table offset metadata, each containing table ID and path to offset file
+ * </ul>
+ *
+ * <p>The actual offset data is stored in remote storage (e.g., OSS/S3) and referenced by the file
+ * paths in {@link TableOffsetMetadata}.
+ *
+ * @see ProducerOffsetsJsonSerde for JSON serialization and deserialization
+ */
+public class ProducerOffsets {
+
+ /** The expiration time in milliseconds since epoch for TTL management. */
+ private final long expirationTime;
+
+ /** List of table offset metadata, each containing table ID and offset file path. */
+ private final List<TableOffsetMetadata> tableOffsets;
+
+ /**
+ * Creates a new ProducerOffsets.
+ *
+ * @param expirationTime the expiration time in milliseconds since epoch
+ * @param tableOffsets list of table offset metadata
+ */
+ public ProducerOffsets(long expirationTime, List<TableOffsetMetadata> tableOffsets) {
+ this.expirationTime = expirationTime;
+ this.tableOffsets = tableOffsets;
+ }
+
+ /**
+ * Returns the expiration time in milliseconds since epoch.
+ *
+ * @return the expiration time
+ */
+ public long getExpirationTime() {
+ return expirationTime;
+ }
+
+ /**
+ * Returns the list of table offset metadata.
+ *
+ * @return list of table offset metadata
+ */
+ public List<TableOffsetMetadata> getTableOffsets() {
+ return tableOffsets;
+ }
+
+ /**
+ * Checks if this snapshot has expired.
+ *
+ * @param currentTimeMs the current time in milliseconds since epoch
+ * @return true if the snapshot has expired, false otherwise
+ */
+ public boolean isExpired(long currentTimeMs) {
+ return currentTimeMs > expirationTime;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof ProducerOffsets)) {
+ return false;
+ }
+ ProducerOffsets that = (ProducerOffsets) o;
+ return expirationTime == that.expirationTime
+ && Objects.equals(tableOffsets, that.tableOffsets);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(expirationTime, tableOffsets);
+ }
+
+ @Override
+ public String toString() {
+ return "ProducerOffsets{"
+ + "expirationTime="
+ + expirationTime
+ + ", tableOffsets="
+ + tableOffsets
+ + '}';
+ }
+
+ /** Metadata for a single table's offset snapshot. */
+ public static class TableOffsetMetadata {
+
+ /** The table ID. */
+ private final long tableId;
+
+ /** The path to the offset file in remote storage. */
+ private final FsPath offsetsPath;
+
+ /**
+ * Creates a new TableOffsetMetadata.
+ *
+ * @param tableId the table ID
+ * @param offsetsPath the path to the offset file
+ */
+ public TableOffsetMetadata(long tableId, FsPath offsetsPath) {
+ this.tableId = tableId;
+ this.offsetsPath = offsetsPath;
+ }
+
+ /**
+ * Returns the table ID.
+ *
+ * @return the table ID
+ */
+ public long getTableId() {
+ return tableId;
+ }
+
+ /**
+ * Returns the path to the offset file.
+ *
+ * @return the offset file path
+ */
+ public FsPath getOffsetsPath() {
+ return offsetsPath;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof TableOffsetMetadata)) {
+ return false;
+ }
+ TableOffsetMetadata that = (TableOffsetMetadata) o;
+ return tableId == that.tableId && Objects.equals(offsetsPath, that.offsetsPath);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(tableId, offsetsPath);
+ }
+
+ @Override
+ public String toString() {
+ return "TableOffsetMetadata{"
+ + "tableId="
+ + tableId
+ + ", offsetsPath="
+ + offsetsPath
+ + '}';
+ }
+ }
+}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/data/producer/ProducerOffsetsJsonSerde.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/producer/ProducerOffsetsJsonSerde.java
new file mode 100644
index 000000000..74b0ed2fb
--- /dev/null
+++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/data/producer/ProducerOffsetsJsonSerde.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.zk.data.producer;
+
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.fluss.utils.json.JsonDeserializer;
+import org.apache.fluss.utils.json.JsonSerializer;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * JSON serializer and deserializer for {@link ProducerOffsets}.
+ *
+ * <p>The serialized format is:
+ *
+ * <pre>{@code
+ * {
+ * "version": 1,
+ * "expiration_time": 1735538268000,
+ * "tables": [
+ * {
+ * "table_id": 100,
+ * "offsets_path": "oss://bucket/path/uuid.offsets"
+ * }
+ * ]
+ * }
+ * }</pre>
+ */
+public class ProducerOffsetsJsonSerde
+ implements JsonSerializer<ProducerOffsets>, JsonDeserializer<ProducerOffsets> {
+
+ public static final ProducerOffsetsJsonSerde INSTANCE = new ProducerOffsetsJsonSerde();
+
+ private static final String VERSION_KEY = "version";
+ private static final String EXPIRATION_TIME_KEY = "expiration_time";
+ private static final String TABLES_KEY = "tables";
+ private static final String TABLE_ID_KEY = "table_id";
+ private static final String OFFSETS_PATH_KEY = "offsets_path";
+
+ private static final int CURRENT_VERSION = 1;
+
+ @Override
+ public void serialize(ProducerOffsets producerOffsets, JsonGenerator generator)
+ throws IOException {
+ generator.writeStartObject();
+
+ generator.writeNumberField(VERSION_KEY, CURRENT_VERSION);
+ generator.writeNumberField(EXPIRATION_TIME_KEY, producerOffsets.getExpirationTime());
+
+ generator.writeArrayFieldStart(TABLES_KEY);
+ for (ProducerOffsets.TableOffsetMetadata tableOffset : producerOffsets.getTableOffsets()) {
+ generator.writeStartObject();
+ generator.writeNumberField(TABLE_ID_KEY, tableOffset.getTableId());
+ generator.writeStringField(OFFSETS_PATH_KEY, tableOffset.getOffsetsPath().toString());
+ generator.writeEndObject();
+ }
+ generator.writeEndArray();
+
+ generator.writeEndObject();
+ }
+
+ @Override
+ public ProducerOffsets deserialize(JsonNode node) {
+ int version = node.get(VERSION_KEY).asInt();
+ if (version != CURRENT_VERSION) {
+ throw new IllegalArgumentException(
+ "Unsupported ProducerOffsets version: "
+ + version
+ + ", expected: "
+ + CURRENT_VERSION);
+ }
+
+ long expirationTime = node.get(EXPIRATION_TIME_KEY).asLong();
+
+ List<ProducerOffsets.TableOffsetMetadata> tableOffsets = new ArrayList<>();
+ JsonNode tablesNode = node.get(TABLES_KEY);
+ if (tablesNode != null && tablesNode.isArray()) {
+ Iterator<JsonNode> elements = tablesNode.elements();
+ while (elements.hasNext()) {
+ JsonNode tableNode = elements.next();
+ long tableId = tableNode.get(TABLE_ID_KEY).asLong();
+ String offsetsPath = tableNode.get(OFFSETS_PATH_KEY).asText();
+ tableOffsets.add(
+ new ProducerOffsets.TableOffsetMetadata(tableId, new FsPath(offsetsPath)));
+ }
+ }
+
+ return new ProducerOffsets(expirationTime, tableOffsets);
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
index 06d15d95e..a0a49e027 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
@@ -52,6 +52,8 @@ import org.apache.fluss.rpc.messages.CreateTableRequest;
import org.apache.fluss.rpc.messages.CreateTableResponse;
import org.apache.fluss.rpc.messages.DatabaseExistsRequest;
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
+import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.DeleteProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
import org.apache.fluss.rpc.messages.DropAclsRequest;
@@ -72,6 +74,8 @@ import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsRequest;
import org.apache.fluss.rpc.messages.GetLatestKvSnapshotsResponse;
import org.apache.fluss.rpc.messages.GetLatestLakeSnapshotRequest;
import org.apache.fluss.rpc.messages.GetLatestLakeSnapshotResponse;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.GetProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.GetTableInfoRequest;
import org.apache.fluss.rpc.messages.GetTableInfoResponse;
import org.apache.fluss.rpc.messages.GetTableSchemaRequest;
@@ -94,6 +98,8 @@ import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotRequest;
import org.apache.fluss.rpc.messages.PrepareLakeTableSnapshotResponse;
import org.apache.fluss.rpc.messages.RebalanceRequest;
import org.apache.fluss.rpc.messages.RebalanceResponse;
+import org.apache.fluss.rpc.messages.RegisterProducerOffsetsRequest;
+import org.apache.fluss.rpc.messages.RegisterProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.RemoveServerTagRequest;
import org.apache.fluss.rpc.messages.RemoveServerTagResponse;
import org.apache.fluss.rpc.messages.TableExistsRequest;
@@ -418,6 +424,24 @@ public class TestCoordinatorGateway implements CoordinatorGateway {
throw new UnsupportedOperationException();
}
+ @Override
+ public CompletableFuture<RegisterProducerOffsetsResponse> registerProducerOffsets(
+ RegisterProducerOffsetsRequest request) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<GetProducerOffsetsResponse> getProducerOffsets(
+ GetProducerOffsetsRequest request) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture<DeleteProducerOffsetsResponse> deleteProducerOffsets(
+ DeleteProducerOffsetsRequest request) {
+ throw new UnsupportedOperationException();
+ }
+
public void setCurrentLeaderEpoch(TableBucket tableBucket, int leaderEpoch) {
currentLeaderEpoch.put(tableBucket, leaderEpoch);
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManagerTest.java
new file mode 100644
index 000000000..68308312c
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsManagerTest.java
@@ -0,0 +1,438 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.coordinator.producer;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.InvalidProducerIdException;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsets;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit tests for {@link ProducerOffsetsManager}. */
+class ProducerOffsetsManagerTest {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(1)
+ .setClusterConf(new Configuration())
+ .build();
+
+ @TempDir Path tempDir;
+
+ private ProducerOffsetsManager manager;
+ private ProducerOffsetsStore store;
+ private ZooKeeperClient zkClient;
+
+ private static final long DEFAULT_TTL_MS = 3600000; // 1 hour
+ private static final long CLEANUP_INTERVAL_MS = 60000; // 1 minute
+
+ @BeforeEach
+ void setUp() {
+ zkClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
+ store = new ProducerOffsetsStore(zkClient, tempDir.toString());
+ manager = new ProducerOffsetsManager(store, DEFAULT_TTL_MS, CLEANUP_INTERVAL_MS);
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (manager != null) {
+ manager.close();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Basic Registration Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRegisterSnapshotSuccess() throws Exception {
+ String producerId = "test-manager-register";
+ Map<TableBucket, Long> offsets = createTestOffsets();
+
+ boolean created = manager.registerSnapshot(producerId, offsets, null);
+
+ assertThat(created).isTrue();
+ Optional<ProducerOffsets> snapshot = manager.getOffsetsMetadata(producerId);
+ assertThat(snapshot).isPresent();
+ assertThat(snapshot.get().getExpirationTime())
+ .isGreaterThanOrEqualTo(System.currentTimeMillis());
+ }
+
+ @Test
+ void testRegisterSnapshotAlreadyExists() throws Exception {
+ String producerId = "test-manager-exists";
+ Map<TableBucket, Long> offsets1 = createTestOffsets();
+ Map<TableBucket, Long> offsets2 = new HashMap<>();
+ offsets2.put(new TableBucket(99L, 0), 999L);
+
+ // First registration should succeed
+ assertThat(manager.registerSnapshot(producerId, offsets1, null)).isTrue();
+
+ // Second registration should return false (already exists)
+ assertThat(manager.registerSnapshot(producerId, offsets2, null)).isFalse();
+
+ // Original offsets should be preserved
+ Map<TableBucket, Long> retrieved = manager.readOffsets(producerId);
+ assertThat(retrieved).containsKey(new TableBucket(1L, 0));
+ assertThat(retrieved).doesNotContainKey(new TableBucket(99L, 0));
+ }
+
+ @Test
+ void testRegisterSnapshotWithCustomTtl() throws Exception {
+ String producerId = "test-manager-custom-ttl";
+ Map<TableBucket, Long> offsets = createTestOffsets();
+ long customTtlMs = 1000; // 1 second
+
+ long beforeRegister = System.currentTimeMillis();
+ manager.registerSnapshot(producerId, offsets, customTtlMs);
+ long afterRegister = System.currentTimeMillis();
+
+ Optional<ProducerOffsets> snapshot = manager.getOffsetsMetadata(producerId);
+ assertThat(snapshot).isPresent();
+ long expirationTime = snapshot.get().getExpirationTime();
+ assertThat(expirationTime).isGreaterThanOrEqualTo(beforeRegister + customTtlMs);
+ assertThat(expirationTime).isLessThanOrEqualTo(afterRegister + customTtlMs);
+ }
+
+ // ------------------------------------------------------------------------
+ // Expired Snapshot Replacement Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRegisterSnapshotReplacesExpired() throws Exception {
+ String producerId = "test-manager-replace-expired";
+ Map<TableBucket, Long> offsets1 = new HashMap<>();
+ offsets1.put(new TableBucket(1L, 0), 100L);
+
+ Map<TableBucket, Long> offsets2 = new HashMap<>();
+ offsets2.put(new TableBucket(2L, 0), 200L);
+
+ // Register with very short TTL (already expired)
+ long expiredTtlMs = -1000; // Already expired
+ store.tryStoreOffsets(producerId, offsets1, System.currentTimeMillis() + expiredTtlMs);
+
+ // Verify it's expired
+ Optional<ProducerOffsets> expiredSnapshot = manager.getOffsetsMetadata(producerId);
+ assertThat(expiredSnapshot).isPresent();
+ assertThat(expiredSnapshot.get().isExpired(System.currentTimeMillis())).isTrue();
+
+ // New registration should replace the expired one
+ boolean created = manager.registerSnapshot(producerId, offsets2, null);
+ assertThat(created).isTrue();
+
+ // Verify new offsets
+ Map<TableBucket, Long> retrieved = manager.readOffsets(producerId);
+ assertThat(retrieved).containsKey(new TableBucket(2L, 0));
+ assertThat(retrieved).doesNotContainKey(new TableBucket(1L, 0));
+ }
+
+ // ------------------------------------------------------------------------
+ // Get and Delete Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testGetOffsetsMetadata() throws Exception {
+ String producerId = "test-manager-get-metadata";
+ Map<TableBucket, Long> offsets = createTestOffsets();
+
+ manager.registerSnapshot(producerId, offsets, null);
+
+ Optional<ProducerOffsets> snapshot = manager.getOffsetsMetadata(producerId);
+ assertThat(snapshot).isPresent();
+ assertThat(snapshot.get().getTableOffsets()).isNotEmpty();
+ }
+
+ @Test
+ void testGetOffsetsMetadataNonExistent() throws Exception {
+ Optional<ProducerOffsets> snapshot = manager.getOffsetsMetadata("non-existent-producer");
+ assertThat(snapshot).isEmpty();
+ }
+
+ @Test
+ void testReadOffsets() throws Exception {
+ String producerId = "test-manager-get-offsets";
+ Map<TableBucket, Long> offsets = createTestOffsets();
+
+ manager.registerSnapshot(producerId, offsets, null);
+
+ Map<TableBucket, Long> retrieved = manager.readOffsets(producerId);
+ assertThat(retrieved).hasSize(offsets.size());
+ assertThat(retrieved.get(new TableBucket(1L, 0))).isEqualTo(100L);
+ }
+
+ @Test
+ void testReadOffsetsNonExistent() throws Exception {
+ Map<TableBucket, Long> offsets = manager.readOffsets("non-existent-producer");
+ assertThat(offsets).isEmpty();
+ }
+
+ @Test
+ void testDeleteSnapshot() throws Exception {
+ String producerId = "test-manager-delete";
+ Map<TableBucket, Long> offsets = createTestOffsets();
+
+ manager.registerSnapshot(producerId, offsets, null);
+ assertThat(manager.getOffsetsMetadata(producerId)).isPresent();
+
+ manager.deleteSnapshot(producerId);
+
+ assertThat(manager.getOffsetsMetadata(producerId)).isEmpty();
+ assertThat(manager.readOffsets(producerId)).isEmpty();
+ }
+
+ @Test
+ void testDeleteNonExistentSnapshot() throws Exception {
+ // Should not throw
+ manager.deleteSnapshot("non-existent-producer");
+ }
+
+ // ------------------------------------------------------------------------
+ // Cleanup Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testCleanupExpiredSnapshots() throws Exception {
+ String expiredProducerId = "test-cleanup-expired";
+ String validProducerId = "test-cleanup-valid";
+
+ // Create an expired snapshot
+ Map<TableBucket, Long> offsets = createTestOffsets();
+ store.tryStoreOffsets(expiredProducerId, offsets, System.currentTimeMillis() - 1000);
+
+ // Create a valid snapshot
+ manager.registerSnapshot(validProducerId, offsets, null);
+
+ // Run cleanup
+ int cleanedCount = manager.cleanupExpiredSnapshots();
+
+ assertThat(cleanedCount).isEqualTo(1);
+ assertThat(manager.getOffsetsMetadata(expiredProducerId)).isEmpty();
+ assertThat(manager.getOffsetsMetadata(validProducerId)).isPresent();
+ }
+
+ @Test
+ void testGetDefaultTtlMs() {
+ assertThat(manager.getDefaultTtlMs()).isEqualTo(DEFAULT_TTL_MS);
+ }
+
+ // ------------------------------------------------------------------------
+ // Producer ID Validation Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testRegisterSnapshotWithInvalidProducerId() {
+ Map<TableBucket, Long> offsets = createTestOffsets();
+
+ // Null producer ID
+ assertThatThrownBy(() -> manager.registerSnapshot(null, offsets, null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID")
+ .hasMessageContaining("null string is not allowed");
+
+ // Empty producer ID
+ assertThatThrownBy(() -> manager.registerSnapshot("", offsets, null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID")
+ .hasMessageContaining("empty string is not allowed");
+
+ // Producer ID with invalid characters
+ assertThatThrownBy(() -> manager.registerSnapshot("invalid/producer", offsets, null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID")
+ .hasMessageContaining("contains one or more characters");
+
+ // Producer ID as "."
+ assertThatThrownBy(() -> manager.registerSnapshot(".", offsets, null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID")
+ .hasMessageContaining("'.' is not allowed");
+
+ // Producer ID as ".."
+ assertThatThrownBy(() -> manager.registerSnapshot("..", offsets, null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID")
+ .hasMessageContaining("'..' is not allowed");
+ }
+
+ @Test
+ void testGetOffsetsMetadataWithInvalidProducerId() {
+ // Null producer ID
+ assertThatThrownBy(() -> manager.getOffsetsMetadata(null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID");
+
+ // Invalid characters
+ assertThatThrownBy(() -> manager.getOffsetsMetadata("invalid@producer"))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID");
+ }
+
+ @Test
+ void testReadOffsetsWithInvalidProducerId() {
+ // Null producer ID
+ assertThatThrownBy(() -> manager.readOffsets(null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID");
+
+ // Invalid characters
+ assertThatThrownBy(() -> manager.readOffsets("invalid producer"))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID");
+ }
+
+ @Test
+ void testDeleteSnapshotWithInvalidProducerId() {
+ // Null producer ID
+ assertThatThrownBy(() -> manager.deleteSnapshot(null))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID");
+
+ // Invalid characters
+ assertThatThrownBy(() -> manager.deleteSnapshot("invalid:producer"))
+ .isInstanceOf(InvalidProducerIdException.class)
+ .hasMessageContaining("Invalid producer ID");
+ }
+
+ // ------------------------------------------------------------------------
+ // Concurrent Registration Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testConcurrentRegistrationSameProducer() throws Exception {
+ String producerId = "test-concurrent-same";
+ int numThreads = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ CountDownLatch doneLatch = new CountDownLatch(numThreads);
+ AtomicInteger createdCount = new AtomicInteger(0);
+ AtomicInteger existsCount = new AtomicInteger(0);
+
+ for (int i = 0; i < numThreads; i++) {
+ final int threadId = i;
+ executor.submit(
+ () -> {
+ try {
+ startLatch.await();
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, threadId), (long) threadId);
+
+ boolean created = manager.registerSnapshot(producerId, offsets, null);
+ if (created) {
+ createdCount.incrementAndGet();
+ } else {
+ existsCount.incrementAndGet();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ } finally {
+ doneLatch.countDown();
+ }
+ });
+ }
+
+ // Start all threads simultaneously
+ startLatch.countDown();
+ doneLatch.await(30, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // Exactly one should succeed with CREATED
+ assertThat(createdCount.get()).isEqualTo(1);
+ assertThat(existsCount.get()).isEqualTo(numThreads - 1);
+
+ // Verify snapshot exists
+ assertThat(manager.getOffsetsMetadata(producerId)).isPresent();
+ }
+
+ @Test
+ void testConcurrentRegistrationDifferentProducers() throws Exception {
+ int numProducers = 10;
+ ExecutorService executor = Executors.newFixedThreadPool(numProducers);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ CountDownLatch doneLatch = new CountDownLatch(numProducers);
+ AtomicInteger successCount = new AtomicInteger(0);
+
+ for (int i = 0; i < numProducers; i++) {
+ final int producerNum = i;
+ executor.submit(
+ () -> {
+ try {
+ startLatch.await();
+ String producerId = "test-concurrent-diff-" + producerNum;
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, producerNum), (long) producerNum);
+
+ boolean created = manager.registerSnapshot(producerId, offsets, null);
+ if (created) {
+ successCount.incrementAndGet();
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ } finally {
+ doneLatch.countDown();
+ }
+ });
+ }
+
+ // Start all threads simultaneously
+ startLatch.countDown();
+ doneLatch.await(30, TimeUnit.SECONDS);
+ executor.shutdown();
+
+ // All should succeed since they're different producers
+ assertThat(successCount.get()).isEqualTo(numProducers);
+
+ // Verify all snapshots exist
+ for (int i = 0; i < numProducers; i++) {
+ assertThat(manager.getOffsetsMetadata("test-concurrent-diff-" + i)).isPresent();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Helper Methods
+ // ------------------------------------------------------------------------
+
+ private Map<TableBucket, Long> createTestOffsets() {
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+ offsets.put(new TableBucket(1L, 1), 200L);
+ offsets.put(new TableBucket(2L, 0), 300L);
+ return offsets;
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsStoreTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsStoreTest.java
new file mode 100644
index 000000000..2386944bc
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/producer/ProducerOffsetsStoreTest.java
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.coordinator.producer;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.server.zk.ZooKeeperClient;
+import org.apache.fluss.server.zk.data.producer.ProducerOffsets;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link ProducerOffsetsStore}. */
+class ProducerOffsetsStoreTest {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(1)
+ .setClusterConf(new Configuration())
+ .build();
+
+ @TempDir Path tempDir;
+
+ private ProducerOffsetsStore store;
+ private ZooKeeperClient zkClient;
+
+ @BeforeEach
+ void setUp() {
+ zkClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
+ store = new ProducerOffsetsStore(zkClient, tempDir.toString());
+ }
+
+ @Test
+ void testTryStoreOffsetsSuccess() throws Exception {
+ String producerId = "test-producer-store";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+ offsets.put(new TableBucket(1L, 1), 200L);
+ offsets.put(new TableBucket(2L, 0), 300L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+
+ // First store should succeed
+ boolean created = store.tryStoreOffsets(producerId, offsets, expirationTime);
+ assertThat(created).isTrue();
+
+ // Verify metadata was stored
+ Optional<ProducerOffsets> snapshot = store.getOffsetsMetadata(producerId);
+ assertThat(snapshot).isPresent();
+ assertThat(snapshot.get().getExpirationTime()).isEqualTo(expirationTime);
+ assertThat(snapshot.get().getTableOffsets()).hasSize(2); // 2 tables
+ }
+
+ @Test
+ void testTryStoreOffsetsAlreadyExists() throws Exception {
+ String producerId = "test-producer-exists";
+ Map<TableBucket, Long> offsets1 = new HashMap<>();
+ offsets1.put(new TableBucket(1L, 0), 100L);
+
+ Map<TableBucket, Long> offsets2 = new HashMap<>();
+ offsets2.put(new TableBucket(1L, 0), 999L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+
+ // First store should succeed
+ assertThat(store.tryStoreOffsets(producerId, offsets1, expirationTime)).isTrue();
+
+ // Second store should return false (already exists)
+ assertThat(store.tryStoreOffsets(producerId, offsets2, expirationTime)).isFalse();
+
+ // Original offsets should be preserved
+ Map<TableBucket, Long> retrieved = store.readOffsets(producerId);
+ assertThat(retrieved).isEqualTo(offsets1);
+ }
+
+ @Test
+ void testReadOffsets() throws Exception {
+ String producerId = "test-producer-read";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+ offsets.put(new TableBucket(1L, 1), 200L);
+ offsets.put(new TableBucket(2L, 0), 300L);
+ offsets.put(new TableBucket(3L, 100L, 0), 400L); // partitioned table
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ // Read offsets back
+ Map<TableBucket, Long> retrieved = store.readOffsets(producerId);
+ assertThat(retrieved).isEqualTo(offsets);
+ }
+
+ @Test
+ void testReadOffsetsNonExistent() throws Exception {
+ Map<TableBucket, Long> offsets = store.readOffsets("non-existent-producer");
+ assertThat(offsets).isEmpty();
+ }
+
+ @Test
+ void testDeleteSnapshot() throws Exception {
+ String producerId = "test-producer-delete";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ // Verify it exists
+ assertThat(store.getOffsetsMetadata(producerId)).isPresent();
+
+ // Delete
+ store.deleteSnapshot(producerId);
+
+ // Verify it's gone
+ assertThat(store.getOffsetsMetadata(producerId)).isEmpty();
+ assertThat(store.readOffsets(producerId)).isEmpty();
+ }
+
+ @Test
+ void testDeleteNonExistentSnapshot() throws Exception {
+ // Should not throw
+ store.deleteSnapshot("non-existent-producer");
+ }
+
+ @Test
+ void testListProducerIds() throws Exception {
+ // Use unique prefix to avoid conflicts with other tests
+ String prefix = "list-test-" + System.currentTimeMillis() + "-";
+
+ // Create multiple snapshots
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+ long expirationTime = System.currentTimeMillis() + 3600000;
+
+ store.tryStoreOffsets(prefix + "producer-1", offsets, expirationTime);
+ store.tryStoreOffsets(prefix + "producer-2", offsets, expirationTime);
+ store.tryStoreOffsets(prefix + "producer-3", offsets, expirationTime);
+
+ List<String> producerIds = store.listProducerIds();
+ assertThat(producerIds)
+ .contains(prefix + "producer-1", prefix + "producer-2", prefix + "producer-3");
+ }
+
+ @Test
+ void testGetProducersDirectory() {
+ FsPath producersDir = store.getProducersDirectory();
+ assertThat(producersDir.toString()).isEqualTo(tempDir.toString() + "/producers");
+ }
+
+ @Test
+ void testSnapshotWithPartitionedTable() throws Exception {
+ String producerId = "test-producer-partitioned";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ // Non-partitioned table
+ offsets.put(new TableBucket(1L, 0), 100L);
+ // Partitioned table with different partitions
+ offsets.put(new TableBucket(2L, 10L, 0), 200L);
+ offsets.put(new TableBucket(2L, 10L, 1), 201L);
+ offsets.put(new TableBucket(2L, 20L, 0), 300L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ Map<TableBucket, Long> retrieved = store.readOffsets(producerId);
+ assertThat(retrieved).isEqualTo(offsets);
+ }
+
+ @Test
+ void testSnapshotExpiration() throws Exception {
+ String producerId = "test-producer-expiration";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+
+ // Create snapshot that expires in the past
+ long pastExpirationTime = System.currentTimeMillis() - 1000;
+ store.tryStoreOffsets(producerId, offsets, pastExpirationTime);
+
+ Optional<ProducerOffsets> snapshot = store.getOffsetsMetadata(producerId);
+ assertThat(snapshot).isPresent();
+ assertThat(snapshot.get().isExpired(System.currentTimeMillis())).isTrue();
+ }
+
+ @Test
+ void testEmptyOffsets() throws Exception {
+ String producerId = "test-producer-empty";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ Optional<ProducerOffsets> snapshot = store.getOffsetsMetadata(producerId);
+ assertThat(snapshot).isPresent();
+ assertThat(snapshot.get().getTableOffsets()).isEmpty();
+
+ Map<TableBucket, Long> retrieved = store.readOffsets(producerId);
+ assertThat(retrieved).isEmpty();
+ }
+
+ // ------------------------------------------------------------------------
+ // Version Control Tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testGetOffsetsMetadataWithVersion() throws Exception {
+ String producerId = "test-producer-version";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ // Get with version
+ Optional<Tuple2<ProducerOffsets, Integer>> result =
+ store.getOffsetsMetadataWithVersion(producerId);
+
+ assertThat(result).isPresent();
+ assertThat(result.get().f0.getExpirationTime()).isEqualTo(expirationTime);
+ assertThat(result.get().f1).isGreaterThanOrEqualTo(0); // ZK version starts at 0
+ }
+
+ @Test
+ void testGetOffsetsMetadataWithVersionNonExistent() throws Exception {
+ Optional<Tuple2<ProducerOffsets, Integer>> result =
+ store.getOffsetsMetadataWithVersion("non-existent-producer");
+
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void testDeleteSnapshotIfVersionSuccess() throws Exception {
+ String producerId = "test-producer-delete-version";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ // Get current version
+ Optional<Tuple2<ProducerOffsets, Integer>> result =
+ store.getOffsetsMetadataWithVersion(producerId);
+ assertThat(result).isPresent();
+ int version = result.get().f1;
+ ProducerOffsets snapshot = result.get().f0;
+
+ // Delete with correct version should succeed
+ boolean deleted = store.deleteSnapshotIfVersion(producerId, snapshot, version);
+ assertThat(deleted).isTrue();
+
+ // Verify deleted
+ assertThat(store.getOffsetsMetadata(producerId)).isEmpty();
+ }
+
+ @Test
+ void testDeleteSnapshotIfVersionMismatch() throws Exception {
+ String producerId = "test-producer-version-mismatch";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ // Get current version
+ Optional<Tuple2<ProducerOffsets, Integer>> result =
+ store.getOffsetsMetadataWithVersion(producerId);
+ assertThat(result).isPresent();
+ ProducerOffsets snapshot = result.get().f0;
+
+ // Delete with wrong version should fail
+ int wrongVersion = 999;
+ boolean deleted = store.deleteSnapshotIfVersion(producerId, snapshot, wrongVersion);
+ assertThat(deleted).isFalse();
+
+ // Snapshot should still exist
+ assertThat(store.getOffsetsMetadata(producerId)).isPresent();
+ }
+
+ @Test
+ void testDeleteSnapshotIfVersionAlreadyDeleted() throws Exception {
+ String producerId = "test-producer-already-deleted";
+ Map<TableBucket, Long> offsets = new HashMap<>();
+ offsets.put(new TableBucket(1L, 0), 100L);
+
+ long expirationTime = System.currentTimeMillis() + 3600000;
+ store.tryStoreOffsets(producerId, offsets, expirationTime);
+
+ // Get snapshot info
+ Optional<Tuple2<ProducerOffsets, Integer>> result =
+ store.getOffsetsMetadataWithVersion(producerId);
+ assertThat(result).isPresent();
+ ProducerOffsets snapshot = result.get().f0;
+
+ // Delete first
+ store.deleteSnapshot(producerId);
+
+ // Delete again with version should return true (already deleted is success)
+ boolean deleted = store.deleteSnapshotIfVersion(producerId, snapshot, 0);
+ assertThat(deleted).isTrue();
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/data/producer/ProducerOffsetsJsonSerdeTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/producer/ProducerOffsetsJsonSerdeTest.java
new file mode 100644
index 000000000..43b5f10c7
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/data/producer/ProducerOffsetsJsonSerdeTest.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.server.zk.data.producer;
+
+import org.apache.fluss.fs.FsPath;
+import org.apache.fluss.utils.json.JsonSerdeTestBase;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+/** Tests for {@link ProducerOffsetsJsonSerde}. */
+class ProducerOffsetsJsonSerdeTest extends JsonSerdeTestBase<ProducerOffsets> {
+
+ ProducerOffsetsJsonSerdeTest() {
+ super(ProducerOffsetsJsonSerde.INSTANCE);
+ }
+
+ @Override
+ protected ProducerOffsets[] createObjects() {
+ // Empty snapshot
+ ProducerOffsets empty = new ProducerOffsets(1735538268000L, Collections.emptyList());
+
+ // Snapshot with multiple tables and different file system schemes
+ ProducerOffsets withTables =
+ new ProducerOffsets(
+ 1735538268000L,
+ Arrays.asList(
+ new ProducerOffsets.TableOffsetMetadata(
+ 100L, new FsPath("oss://bucket/path/uuid1.offsets")),
+ new ProducerOffsets.TableOffsetMetadata(
+ 200L, new FsPath("s3://bucket/path/uuid2.offsets"))));
+
+ return new ProducerOffsets[] {empty, withTables};
+ }
+
+ @Override
+ protected String[] expectedJsons() {
+ return new String[] {
+ "{\"version\":1,\"expiration_time\":1735538268000,\"tables\":[]}",
+ "{\"version\":1,\"expiration_time\":1735538268000,\"tables\":["
+ + "{\"table_id\":100,\"offsets_path\":\"oss://bucket/path/uuid1.offsets\"},"
+ + "{\"table_id\":200,\"offsets_path\":\"s3://bucket/path/uuid2.offsets\"}]}"
+ };
+ }
+}
diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md
index 44d6e6278..93630a969 100644
--- a/website/docs/maintenance/configuration.md
+++ b/website/docs/maintenance/configuration.md
@@ -53,9 +53,11 @@ during the Fluss cluster working.
## CoordinatorServer
-| Option | Type | Default | Description |
-|--------------------------|---------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| coordinator.io-pool.size | Integer | 10 | **Deprecated**: This option is deprecated. Please use `server.io-pool.size` instead. The size of the IO thread pool to run blocking operations for coordinator server. This includes discard unnecessary snapshot files. Increase this value if you experience slow unnecessary snapshot files clean. The default value is 10. |
+| Option | Type | Default | Description |
+|----------------------------------------------|----------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| coordinator.io-pool.size | Integer | 10 | **Deprecated**: This option is deprecated. Please use `server.io-pool.size` instead. The size of the IO thread pool to run blocking operations for coordinator server. This includes discard unnecessary snapshot files. Increase this value if you experience slow unnecessary snapshot files clean. The default value is 10. |
+| coordinator.producer-offsets.ttl | Duration | 24h | The TTL (time-to-live) for producer offsets. Producer offsets older than this TTL will be automatically cleaned up by the coordinator server. Producer offsets are used for undo recovery when a Flink job fails over before completing its first checkpoint. The default value is 24 hours. |
+| coordinator.producer-offsets.cleanup-interval | Duration | 1h | The interval for cleaning up expired producer offsets and orphan files in remote storage. The cleanup task runs periodically to remove expired offsets and any orphan files that may have been left behind due to incomplete operations. The default value is 1 hour. |
## TabletServer