Skip to content

[FLUSS] Add Producer Offset Snapshot for Exactly-Once semantics #2434

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Open
platinumhamburg wants to merge 7 commits into
base: main
Choose a base branch
from
Open

[FLUSS] Add Producer Offset Snapshot for Exactly-Once semantics #2434

platinumhamburg wants to merge 7 commits into from
+4,125 −22

Conversation

@platinumhamburg
Copy link
Contributor

@platinumhamburg platinumhamburg commented Jan 22, 2026

Core changes:

  • Add ProducerSnapshotManager for lifecycle management with atomic registration
  • Add ProducerSnapshotStore for ZK + remote storage operations
  • Add tryRegisterProducerSnapshot in ZooKeeperClient for atomic check-and-create
  • Add Admin API: registerProducerOffsets, getProducerOffsets, deleteProducerOffsets
  • Add configurable TTL and cleanup interval for producer snapshots

Design highlights:

  • Atomic registration via ZK's NodeExistsException handling
  • Eventually consistent: ZK as commit point, orphan files cleaned periodically
  • UUID-based file naming prevents concurrent upload conflicts

Tests:

  • ProducerSnapshotManagerTest: lifecycle, expiration, concurrent atomicity
  • ProducerSnapshotJsonSerdeTest: JSON format compatibility

Purpose

Linked issue: close #2433

Brief change log

Tests

API and Format

Documentation

@platinumhamburg
[FLUSS] Add Producer Offset Snapshot for Exactly-Once Semantics
478e8ec 
This commit introduces the Producer Offset Snapshot feature to support
exactly-once semantics in Fluss. The feature allows producers (e.g., Flink
jobs) to register their offset snapshots for recovery purposes.

Main changes:
- Add ProducerSnapshotManager for lifecycle management of producer snapshots
- Add ProducerSnapshotStore for low-level storage operations (ZK + remote FS)
- Add Admin APIs: registerProducerOffsets, getProducerOffsets, deleteProducerOffsets
- Add RetryUtils for IO operations with exponential backoff
- Add configuration options for snapshot TTL and cleanup interval

Code quality improvements:
- Remove ProducerSnapshotResultCodes class, use RegisterResult enum directly
- Fix RetryUtils interrupt handling to preserve thread interrupt status
- Add comprehensive tests for ProducerSnapshotManager including concurrency tests
- Add interrupt handling tests for RetryUtils
wuchong
wuchong reviewed Jan 25, 2026
View reviewed changes
Copy link
Member

@wuchong wuchong left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @platinumhamburg , I left some comments.

@platinumhamburg
[FLUSS] Refactor producer offsets management and improve API
0ee999d 
- Rename 'producer-snapshot' config options to 'producer-offsets' for clarity
- Refactor RegisterProducerOffsetsRequest to use PbProducerTableOffsets
  instead of flat PbTableBucketOffset for better organization by table
- Move RegisterResult enum from fluss-rpc to fluss-client as public API
- Add producer ID validation using TablePath.detectInvalidName()
- Improve authorization: check per-table permissions instead of cluster-level
- Add version-based delete in cleanup to avoid race conditions
- Use FlussPaths for consistent remote storage path generation
- Add FileNotFoundException handling for non-retryable file errors
- Extract RPC message conversion utilities to ClientRpcMessageUtils
  and ServerRpcMessageUtils
- Add comprehensive unit tests for producer ID validation
@platinumhamburg
Rename producer snapshot classes to producer offsets
41d103d 
Rename classes and corresponding variable declarations:
- ProducerSnapshot -> ProducerOffsets
- ProducerSnapshotJsonSerde -> ProducerOffsetsJsonSerde
- ProducerSnapshotStore -> ProducerOffsetsStore
- ProducerSnapshotManager -> ProducerOffsetsManager

Update variable names in:
- CoordinatorService: producerSnapshotManager -> producerOffsetsManager
- ZooKeeperClient: class type references updated
- ZkData: class type references updated

Test classes renamed accordingly.
@platinumhamburg
Copy link
Contributor Author

platinumhamburg commented Jan 26, 2026

Thanks @wuchong for the detailed review. I’ve addressed all the comments above, please take another look when you have time.

wuchong
wuchong reviewed Jan 26, 2026
View reviewed changes
Copy link
Member

@wuchong wuchong left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@platinumhamburg the updated changes looks good to me in general. I left some final comments. Besides, I appended a commit to improve renaming and comments a bit.

@platinumhamburg
[FLUSS-811924] Improve producer offsets API exception handling
7c294fc 
- Remove duplicated RegisterResult class from fluss-common, use magic values
  (0=CREATED, 1=ALREADY_EXISTS) with comments aligning to client RegisterResult enum
- Introduce InvalidProducerIdException (ApiException) for producer ID validation
- Register InvalidProducerIdException in Errors enum (code 63)
- Re-throw ApiExceptions as-is in producer offsets APIs to preserve exception types
- Wrap non-ApiExceptions with UnknownServerException instead of RuntimeException
- Add authorization tests for producer offsets operations in FlussAuthorizationITCase
@platinumhamburg
Copy link
Contributor Author

platinumhamburg commented Jan 27, 2026

@wuchong Thanks for your detailed review and suggestions. I’ve addressed all the comments above—please take a look when you have time.

@platinumhamburg
[fluss-server] Fix authorization in async producer offsets operations
f80d940 
The deleteProducerOffsets and getProducerOffsets methods were calling
authorizeTable() inside CompletableFuture.supplyAsync(), which runs on
a different thread (ioExecutor). Since currentSession() relies on
thread-local storage, it fails with 'No session set' error when called
from the async thread.

Fix:
- Capture session before entering async block
- Add authorizeTableWithSession() method that accepts explicit session
- Refactor authorizeTable(OperationType, long) to delegate to the new
  method, eliminating code duplication
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@wuchong wuchong wuchong approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

[Server] Add Producer Offset Snapshot Registry as Infrastructure for Exactly-Once Semantics

2 participants

@platinumhamburg @wuchong