Fix Kafka change feed continuation after partition splits by tvaron3 · Pull Request #50031 · Azure/azure-sdk-for-java

Tomas Varon (tvaron3) · 2026-08-04T21:23:27Z

Summary

Project stale parent change feed continuations onto each post-split child range.
Preserve record-level LSN overrides only for the token range that produced them.
Apply the same projection during runtime split handling and connector restart/rebalance.
Fail loudly when split resolution returns no overlapping ranges instead of silently dropping the task.
Add Core, Kafka state, restart, and live source-task regression coverage.

Why

A parent continuation can contain divergent child LSNs after a physical partition split. Kafka previously collapsed that state to one scalar token and applied it to every child, which could advance a lower-LSN child past unread records. A live repro confirmed permanent marker loss. This is related to the Spark bounded change feed issue in #49883, but Kafka manifests the shared continuation-projection mistake as data loss rather than a bounded-read hang.

Project composite parent continuations onto each target feed range so sibling LSNs cannot be reused after a partition split. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3308d1b4-b9e6-4c57-b44f-82a6331ab010

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3308d1b4-b9e6-4c57-b44f-82a6331ab010

azure-pipelines · 2026-08-04T21:24:30Z

Azure Pipelines: Successfully started running 2 pipeline(s). 32 pipeline(s) were filtered out due to trigger conditions. There may be pipelines that require an authorized user to comment /azp run to run.

azure-pipelines · 2026-08-04T22:38:28Z

Azure Pipelines: Successfully started running 2 pipeline(s). 32 pipeline(s) were filtered out due to trigger conditions. There may be pipelines that require an authorized user to comment /azp run to run.

Copilot

Pull request overview

Fixes Kafka Cosmos DB change feed continuation handling after physical partition splits by projecting stale parent continuations onto the effective child feed ranges, preventing one child’s higher LSN from incorrectly advancing a sibling range (data loss risk). This mirrors the underlying “composite continuation must not be collapsed” lesson from the related Spark fix (#49883), but applies it to the Kafka connector’s runtime split handling and restart/rebalance paths.

Changes:

Add continuation projection utilities in CosmosChangeFeedRequestOptions to scope composite continuation tokens (and item-level LSN overrides) to the correct effective feed range.
Update Kafka connector state handling to project/partition continuations on restart/rebalance and when resolving splits/merges; fail fast when split resolution returns no overlaps.
Add unit + connector regression tests covering divergent child continuations, state extraction behavior, and a live source-task scenario.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/models/CosmosChangeFeedRequestOptions.java	Implements continuation projection to an effective `FeedRange` and exposes helper to extract range-scoped continuation strings.
sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java	Extends the internal accessor to allow extracting a feed-range-scoped continuation.
sdk/cosmos/azure-cosmos/CHANGELOG.md	Adds release note for the Kafka continuation projection fix.
sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/KafkaChangeFeedContinuationProjectionTest.java	Adds unit tests for projecting parent continuation state onto child ranges and scoping item-LSN overrides.
sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedStateTest.java	Adds tests ensuring `KafkaCosmosChangeFeedState` extraction scopes continuation correctly and preserves/drops item LSN as intended.
sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTaskTest.java	Adds a regression test that exercises polling with divergent child continuations to ensure no missed marker after projection.
sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSourceConnectorTest.java	Updates split/restart test setup to use child-scoped continuation tokens and to project state per feed range when building task units.
sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedState.java	Adds extraction method that projects continuation to a target feed range and scopes item-LSN override to the owning range; improves null itemLsn deserialization.
sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java	Uses projected child continuation state for new ranges and fails loudly when no overlaps are found.
sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSourceConnector.java	Projects offset continuation state onto current container feed ranges for restart/rebalance correctness.
sdk/cosmos/azure-cosmos-kafka-connect/pom.xml	Exports routing internals needed by new tests/utilities under JPMS test runs.
sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md	Adds release note for the connector-side data loss fix after splits.

Tomas Varon (tvaron3) · 2026-08-04T22:51:54Z

/azp run java - cosmos - kafka

azure-pipelines · 2026-08-04T22:52:12Z

Azure Pipelines: Successfully started running 1 pipeline(s).

Annie Liang (xinlian12) · 2026-08-11T19:42:46Z

@sdkReviewAgent

Fabian Meiswinkel (FabianMeiswinkel)

LGTM - Thx

Annie Liang (xinlian12) · 2026-08-11T21:13:33Z

+            .CosmosChangeFeedRequestOptionsHelper
+            .getCosmosChangeFeedRequestOptionsAccessor()
+            .extractContinuationForFeedRange(this.responseContinuation, feedRange);
+        String projectedItemLsn = this.targetRange.equals(feedRange) ? this.itemLsn : null;


🔴 Correctness: Preserve the record rollback when projecting to children

String projectedItemLsn = this.targetRange.equals(feedRange) ? this.itemLsn : null;

Each emitted record stores the page-end continuation together with that record's LSN. If Kafka commits only part of a page and the parent then splits, the parent range differs from every child, so this clears the rollback LSN from all children. The producing child consequently resumes at the page-end token and permanently skips the uncommitted suffix of that page.

Please project the continuation and LSN atomically, retaining the override for the original token's overlapping child range(s), rather than deciding ownership from overall target-range equality.

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.

Annie Liang (xinlian12) · 2026-08-11T21:13:33Z

-                if (overlappedRanges.size() == 1) {
+                if (overlappedRanges.isEmpty()) {
+                    return Mono.error(
+                        new IllegalStateException(


🟡 Error Handling: Keep empty routing resolution retriable

new IllegalStateException(

CosmosAsyncContainer.getOverlappingFeedRanges also maps a temporarily unavailable/null routing map to an empty list. This exception later becomes a plain ConnectException, because only transient CosmosExceptions are converted to RetriableException, so a cache-refresh failure permanently stops the task instead of retrying.

Failing loudly is correct, but this path should preserve retryability (or retry the routing refresh explicitly) so transient metadata state does not require manual task recovery.

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.

Annie Liang (xinlian12) · 2026-08-11T21:13:33Z

-                );
-
-            return createForProcessingFromContinuation(targetChangeFeedState);
+        CompositeContinuationToken parentCurrentToken = parentContinuation.getCurrentContinuationToken();


🔴 Correctness: The current token is not the token that produced the record

CompositeContinuationToken parentCurrentToken = parentContinuation.getCurrentContinuationToken();

For a non-empty page, ChangeFeedFetcher updates the producing token and then moveToNextToken() rotates it to the queue tail before Fetcher writes the transformed continuation back to the response. Kafka stores that post-rotation continuation with each record's LSN, so on resume the current/head token is the next range, not the producing range. Applying the LSN here can advance an unread sibling while leaving the producer at page end, losing records in both ranges.

Please persist the producing effective range with the offset (or otherwise preserve its identity through rotation) and add a lifecycle test built from an actual transformed response continuation rather than a manually ordered token list.

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.

Annie Liang (xinlian12) · 2026-08-11T21:13:33Z

-
-            return createForProcessingFromContinuation(targetChangeFeedState);
+        CompositeContinuationToken parentCurrentToken = parentContinuation.getCurrentContinuationToken();
+        boolean overlapsTarget = parentState.extractContinuationTokens().stream()


🟡 Continuation Validation: Require complete target coverage

boolean overlapsTarget = parentState.extractContinuationTokens().stream()

Checking for any overlap is not sufficient for a fail-safe projection. For example, a target [00, FF) with saved tokens covering only [00, 80) passes this check; extractForEffectiveRange then returns the clipped state and the task silently stops covering [80, FF). Gapped token sets are accepted similarly.

Please validate that the projected tokens form an exact contiguous tiling of the normalized target (first min, every adjacent boundary, and final max) before constructing request options.

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.

Annie Liang (xinlian12) · 2026-08-11T21:13:33Z

+                    partitionKeyForRange("marker-b", childB, containerProperties.getPartitionKeyDefinition()),
+                    "marker-b")).block();
+
+            FeedRangeTaskUnit feedRangeTaskUnit = new FeedRangeTaskUnit(


🟡 Tests: Exercise the runtime split boundary

FeedRangeTaskUnit feedRangeTaskUnit = new FeedRangeTaskUnit(

This starts directly with a child-scoped task carrying the parent continuation. It validates Core projection during request creation, but it never executes the changed handleFeedRangeGone → getChildRangeChangeFeedState fan-out path. A regression there back to scalar cloning would therefore leave this live test green.

Please drive a stale parent task through a real or injected 410/split resolution, then assert the generated child tasks retain their distinct tokens and the lower-LSN child's marker is delivered.

⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.

Annie Liang (xinlian12) · 2026-08-11T21:13:40Z

✅ Review complete (09:54)

Posted 5 inline comment(s).

_{Steps: ✓ context, correctness, cross-sdk, design, history, past-prs, synthesis, test-coverage}

Annie Liang (xinlian12) · 2026-08-11T22:03:00Z

@@ -478,7 +482,7 @@ private KafkaCosmosChangeFeedState getChildRangeChangeFeedState(
        KafkaCosmosChangeFeedState parent,
        FeedRange feedRange) {
        return parent == null
-            ? null : new KafkaCosmosChangeFeedState(parent.getResponseContinuation(), feedRange);
+            ? null : parent.extractForFeedRange(feedRange);


after reading and thinking more, I think the changes in this PR made the code more robust, but the loss partition data after split should not exist today - the main reason is because in kafka we disabled the split handling in the underlying azure-cosmos sdk, which means for each KafkaChangeFeedState, it should always maintain a 1:1 mapping of the feedRange and continuationToken.

Annie Liang (xinlian12) · 2026-08-11T22:03:19Z

@@ -481,11 +482,14 @@ private Mono<Map<FeedRange, KafkaCosmosChangeFeedState>> getEffectiveContinuatio
                            );

                            if (continuationTopicOffset == null) {
-                                effectiveContinuationMap.put(overlappedFeedRangesFromOffset.get(0), null);
+                                effectiveContinuationMap.put(containerFeedRange, null);


this change will help the scalability after split~

github-actions Bot added the Cosmos label Aug 4, 2026

Link Kafka split fix changelogs

6cdf1e3

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 3308d1b4-b9e6-4c57-b44f-82a6331ab010

Tomas Varon (tvaron3) marked this pull request as ready for review August 4, 2026 22:37

Tomas Varon (tvaron3) requested a review from Kiran Kumar Kolli (kirankumarkolli) as a code owner August 4, 2026 22:37

Copilot AI lite review requested due to automatic review settings August 4, 2026 22:37

Tomas Varon (tvaron3) requested review from a team as code owners August 4, 2026 22:37

Copilot started reviewing on behalf of Tomas Varon (tvaron3) August 4, 2026 22:38 View session

Copilot AI reviewed Aug 4, 2026

View reviewed changes

Comment thread ...ect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java

Comment thread sdk/cosmos/azure-cosmos/CHANGELOG.md

Fabian Meiswinkel (FabianMeiswinkel) approved these changes Aug 11, 2026

View reviewed changes

Annie Liang (xinlian12) reviewed Aug 11, 2026

View reviewed changes

Tomas Varon (tvaron3) closed this Aug 11, 2026

Uh oh!

Conversation

Tomas Varon (tvaron3) commented Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Why

Uh oh!

azure-pipelines Bot commented Aug 4, 2026

Uh oh!

azure-pipelines Bot commented Aug 4, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Tomas Varon (tvaron3) commented Aug 4, 2026

Uh oh!

azure-pipelines Bot commented Aug 4, 2026

Uh oh!

Annie Liang (xinlian12) commented Aug 11, 2026

Uh oh!

Fabian Meiswinkel (FabianMeiswinkel) left a comment

Choose a reason for hiding this comment

Uh oh!

Annie Liang (xinlian12) Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Annie Liang (xinlian12) Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Annie Liang (xinlian12) Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Annie Liang (xinlian12) Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Annie Liang (xinlian12) Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Annie Liang (xinlian12) commented Aug 11, 2026

Uh oh!

Annie Liang (xinlian12) Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Annie Liang (xinlian12) Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Tomas Varon (tvaron3) commented Aug 4, 2026 •

edited

Loading