You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "platinumhamburg (via GitHub)" <gi...@apache.org> on 2026/01/30 12:25:24 UTC
[PR] [FLUSS-CLIENT] Return log end offset in UpsertResult and DeleteResult for exactly-once semantics [fluss]
platinumhamburg opened a new pull request, #2530:
URL: https://github.com/apache/fluss/pull/2530
This change enhances the UpsertWriter API to return bucket and log end offset information in UpsertResult and DeleteResult, which is essential for implementing exactly-once semantics in Flink sink connectors.
Changes:
- Add bucket and logEndOffset fields to UpsertResult and DeleteResult
- Extend WriteCallback interface to include bucket and logEndOffset parameters
- Add log_end_offset field to PbPutKvRespForBucket protobuf message
- Update Sender to propagate log end offset from server response to callbacks
- Add sendWithResult() method in AbstractTableWriter for result-aware writes
- Add integration tests for log end offset tracking
The log end offset (LEO) represents the offset of the next record to be written, allowing Flink sink to track checkpoint offsets for exactly-once delivery.
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #2529 2529
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLUSS-CLIENT] Return log end offset in UpsertResult and DeleteResult for exactly-once semantics [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #2530:
URL: https://github.com/apache/fluss/pull/2530#discussion_r2751139359
##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java:
##########
@@ -1805,4 +1807,108 @@ void testPkCompactedPollFromLatestNoRecords() throws Exception {
scanner.close();
}
}
+
+ // ---------------------- Upsert/Delete Result with LogEndOffset tests ----------------------
+
+ @Test
+ void testUpsertAndDeleteReturnLogEndOffset() throws Exception {
+ // Create a PK table with single bucket for predictable offset tracking
+ TablePath tablePath = TablePath.of("test_db_1", "test_upsert_delete_log_end_offset");
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.STRING())
+ .primaryKey("a")
+ .build();
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(schema).distributedBy(1, "a").build();
+ createTable(tablePath, tableDescriptor, true);
+
+ try (Table table = conn.getTable(tablePath)) {
+ UpsertWriter upsertWriter = table.newUpsert().createWriter();
+ TableBucket expectedBucket = new TableBucket(table.getTableInfo().getTableId(), 0);
+
+ // First upsert - should return log end offset > 0
+ UpsertResult upsertResult1 = upsertWriter.upsert(row(1, "a")).get();
+ assertThat(upsertResult1.getBucket()).isEqualTo(expectedBucket);
+ assertThat(upsertResult1.getLogEndOffset()).isGreaterThan(0);
Review Comment:
Since there is only one bucket, I think we can assert the log end offset with specific value. This ensures the correctness of the end offset.
##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java:
##########
@@ -1805,4 +1807,108 @@ void testPkCompactedPollFromLatestNoRecords() throws Exception {
scanner.close();
}
}
+
+ // ---------------------- Upsert/Delete Result with LogEndOffset tests ----------------------
+
+ @Test
+ void testUpsertAndDeleteReturnLogEndOffset() throws Exception {
+ // Create a PK table with single bucket for predictable offset tracking
+ TablePath tablePath = TablePath.of("test_db_1", "test_upsert_delete_log_end_offset");
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.STRING())
+ .primaryKey("a")
+ .build();
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(schema).distributedBy(1, "a").build();
+ createTable(tablePath, tableDescriptor, true);
+
+ try (Table table = conn.getTable(tablePath)) {
+ UpsertWriter upsertWriter = table.newUpsert().createWriter();
+ TableBucket expectedBucket = new TableBucket(table.getTableInfo().getTableId(), 0);
+
+ // First upsert - should return log end offset > 0
+ UpsertResult upsertResult1 = upsertWriter.upsert(row(1, "a")).get();
+ assertThat(upsertResult1.getBucket()).isEqualTo(expectedBucket);
+ assertThat(upsertResult1.getLogEndOffset()).isGreaterThan(0);
+ long firstOffset = upsertResult1.getLogEndOffset();
+
+ // Second upsert - should return higher log end offset
+ UpsertResult upsertResult2 = upsertWriter.upsert(row(2, "b")).get();
+ assertThat(upsertResult2.getBucket()).isEqualTo(expectedBucket);
+ assertThat(upsertResult2.getLogEndOffset()).isGreaterThan(firstOffset);
+ long secondOffset = upsertResult2.getLogEndOffset();
+
+ // Update existing key - should return higher log end offset
+ UpsertResult upsertResult3 = upsertWriter.upsert(row(1, "aa")).get();
+ assertThat(upsertResult3.getBucket()).isEqualTo(expectedBucket);
+ assertThat(upsertResult3.getLogEndOffset()).isGreaterThan(secondOffset);
+ long thirdOffset = upsertResult3.getLogEndOffset();
+
+ // Delete - should return higher log end offset
+ DeleteResult deleteResult = upsertWriter.delete(row(1, "aa")).get();
+ assertThat(deleteResult.getBucket()).isEqualTo(expectedBucket);
+ assertThat(deleteResult.getLogEndOffset()).isGreaterThan(thirdOffset);
+
+ // Verify the data via lookup
+ Lookuper lookuper = table.newLookup().createLookuper();
+ // key 1 should be deleted
+ assertThat(lookupRow(lookuper, row(1))).isNull();
+ // key 2 should exist
+ assertThat(lookupRow(lookuper, row(2))).isNotNull();
+ }
+ }
+
+ @Test
+ void testBatchedUpsertReturnsSameLogEndOffset() throws Exception {
+ // Test that multiple records in the same batch receive the same log end offset
+ TablePath tablePath = TablePath.of("test_db_1", "test_batched_upsert_log_end_offset");
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("b", DataTypes.STRING())
+ .primaryKey("a")
+ .build();
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(schema).distributedBy(1, "a").build();
+ createTable(tablePath, tableDescriptor, true);
+
+ // Configure small batch size to ensure records are batched together
+ Configuration config = new Configuration(clientConf);
+ config.set(ConfigOptions.CLIENT_WRITER_BATCH_SIZE, new MemorySize(1024 * 1024)); // 1MB
+ config.set(
+ ConfigOptions.CLIENT_WRITER_MAX_INFLIGHT_REQUESTS_PER_BUCKET, 1); // Force batching
+
+ try (Connection connection = ConnectionFactory.createConnection(config);
+ Table table = connection.getTable(tablePath)) {
+ UpsertWriter upsertWriter = table.newUpsert().createWriter();
+
+ // Send multiple upserts without waiting - they should be batched
+ CompletableFuture<UpsertResult> future1 = upsertWriter.upsert(row(1, "a"));
+ CompletableFuture<UpsertResult> future2 = upsertWriter.upsert(row(2, "b"));
+ CompletableFuture<UpsertResult> future3 = upsertWriter.upsert(row(3, "c"));
+
+ // Flush to send the batch
+ upsertWriter.flush();
+
+ // Get results
+ UpsertResult result1 = future1.get();
+ UpsertResult result2 = future2.get();
+ UpsertResult result3 = future3.get();
+
+ // All results should have valid bucket and log end offset
+ assertThat(result1.getBucket()).isNotNull();
+ assertThat(result1.getLogEndOffset()).isGreaterThan(0);
+ assertThat(result2.getBucket()).isNotNull();
+ assertThat(result2.getLogEndOffset()).isGreaterThan(0);
+ assertThat(result3.getBucket()).isNotNull();
+ assertThat(result3.getLogEndOffset()).isGreaterThan(0);
Review Comment:
ditto
##########
fluss-client/src/main/java/org/apache/fluss/client/write/WriteBatch.java:
##########
@@ -176,11 +183,18 @@ public long getQueueTimeMs() {
return drainedMs - createdMs;
}
- /** Complete the batch successfully. */
+ /** Complete the batch successfully (for log batches without offset tracking). */
public boolean complete() {
return done(null);
}
+ /** Complete the batch successfully with bucket and log end offset info (for KV batches). */
+ public boolean complete(TableBucket bucket, long logEndOffset) {
+ this.resultBucket = bucket;
+ this.resultLogEndOffset = logEndOffset;
+ return done(null);
Review Comment:
I think we can make `bucket ` and `logEndOffset` being parameters of `done(..)` and `completeFutureAndFireCallbacks(..)`, just like how we handle the `exception` parameter. This avoid member variables `resultBucket` and `resultLogEndOffset` which looks like having concurrency issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLUSS-CLIENT] Return log end offset in UpsertResult and DeleteResult for exactly-once semantics [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #2530:
URL: https://github.com/apache/fluss/pull/2530
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org