diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md index 65d99b6cf6fa..7daa7cdc71c4 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md @@ -7,6 +7,7 @@ #### Breaking Changes #### Bugs Fixed +* Fixed source connector data loss after partition splits when a stale parent continuation contained divergent child LSNs. - See [PR 50031](https://github.com/Azure/azure-sdk-for-java/pull/50031) #### Other Changes @@ -132,4 +133,3 @@ * Added `ServicePrincipal` support - See [PR 39490](https://github.com/Azure/azure-sdk-for-java/pull/39490) * Added `ItemPatch support` in sink connector - See [PR 39558](https://github.com/Azure/azure-sdk-for-java/pull/39558) * Added support to use CosmosDB container for tracking metadata - See [PR 39634](https://github.com/Azure/azure-sdk-for-java/pull/39634) - diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml b/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml index 1b794b7b9aa4..520de3e9e517 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml +++ b/sdk/cosmos/azure-cosmos-kafka-connect/pom.xml @@ -62,6 +62,7 @@ Licensed under the MIT License. --add-exports com.azure.cosmos/com.azure.cosmos.implementation.changefeed.common=com.azure.cosmos.kafka.connect --add-exports com.azure.cosmos/com.azure.cosmos.implementation.feedranges=com.azure.cosmos.kafka.connect --add-exports com.azure.cosmos/com.azure.cosmos.implementation.query=com.azure.cosmos.kafka.connect + --add-exports com.azure.cosmos/com.azure.cosmos.implementation.routing=com.azure.cosmos.kafka.connect - diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSourceConnector.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSourceConnector.java index 02ee4dce1105..2ba3c767a597 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSourceConnector.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/CosmosSourceConnector.java @@ -447,6 +447,7 @@ private Mono> getEffectiveContinuatio containerFeedRange, this.getContinuationStateFromOffset( feedRangeContinuationTopicOffset, + containerFeedRange, containerFeedRange)); return Mono.just(effectiveContinuationMap); @@ -481,11 +482,14 @@ private Mono> getEffectiveContinuatio ); if (continuationTopicOffset == null) { - effectiveContinuationMap.put(overlappedFeedRangesFromOffset.get(0), null); + effectiveContinuationMap.put(containerFeedRange, null); } else { effectiveContinuationMap.put( containerFeedRange, - this.getContinuationStateFromOffset(continuationTopicOffset, containerFeedRange)); + this.getContinuationStateFromOffset( + continuationTopicOffset, + overlappedFeedRangesFromOffset.get(0), + containerFeedRange)); } return Mono.just(effectiveContinuationMap); @@ -507,6 +511,7 @@ private Mono> getEffectiveContinuatio overlappedRangeFromOffset, this.getContinuationStateFromOffset( this.kafkaOffsetStorageReader.getFeedRangeContinuationOffset(databaseName, containerRid, overlappedRangeFromOffset), + overlappedRangeFromOffset, overlappedRangeFromOffset)); } } @@ -522,15 +527,16 @@ private Mono> getEffectiveContinuatio private KafkaCosmosChangeFeedState getContinuationStateFromOffset( FeedRangeContinuationTopicOffset feedRangeContinuationTopicOffset, - FeedRange feedRange) { + FeedRange offsetFeedRange, + FeedRange targetFeedRange) { - KafkaCosmosChangeFeedState changeFeedState = + KafkaCosmosChangeFeedState offsetState = new KafkaCosmosChangeFeedState( feedRangeContinuationTopicOffset.getResponseContinuation(), - feedRange, + offsetFeedRange, feedRangeContinuationTopicOffset.getItemLsn()); - return changeFeedState; + return offsetState.extractForFeedRange(targetFeedRange); } private List getFeedRanges(CosmosContainerProperties containerProperties) { diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java index 7c7fd12fb059..5f88812cb81d 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTask.java @@ -407,7 +407,11 @@ private Mono handleFeedRangeGone(FeedRangeTaskUnit feedRangeTaskUnit) { .getOverlappingFeedRanges(container, feedRangeTaskUnit.getFeedRange(), true) .flatMap(overlappedRanges -> { - if (overlappedRanges.size() == 1) { + if (overlappedRanges.isEmpty()) { + return Mono.error( + new IllegalStateException( + "No overlapping feed ranges found for " + feedRangeTaskUnit.getFeedRange())); + } else if (overlappedRanges.size() == 1) { // merge happens LOGGER.info( "FeedRange {} is merged into {}, but we will continue polling data from feedRange {}", @@ -478,7 +482,7 @@ private KafkaCosmosChangeFeedState getChildRangeChangeFeedState( KafkaCosmosChangeFeedState parent, FeedRange feedRange) { return parent == null - ? null : new KafkaCosmosChangeFeedState(parent.getResponseContinuation(), feedRange); + ? null : parent.extractForFeedRange(feedRange); } private CosmosChangeFeedRequestOptions getChangeFeedRequestOptions(FeedRangeTaskUnit feedRangeTaskUnit) { diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedState.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedState.java index e5e0fe89a8a1..75fbb757a9d0 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedState.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedState.java @@ -3,6 +3,7 @@ package com.azure.cosmos.kafka.connect.implementation.source; +import com.azure.cosmos.implementation.ImplementationBridgeHelpers; import com.azure.cosmos.implementation.apachecommons.lang.StringUtils; import com.azure.cosmos.models.FeedRange; import com.fasterxml.jackson.core.JsonGenerator; @@ -52,6 +53,18 @@ public String getItemLsn() { return itemLsn; } + public KafkaCosmosChangeFeedState extractForFeedRange(FeedRange feedRange) { + checkNotNull(feedRange, "Argument 'feedRange' can not be null"); + + String projectedContinuation = ImplementationBridgeHelpers + .CosmosChangeFeedRequestOptionsHelper + .getCosmosChangeFeedRequestOptionsAccessor() + .extractContinuationForFeedRange(this.responseContinuation, feedRange); + String projectedItemLsn = this.targetRange.equals(feedRange) ? this.itemLsn : null; + + return new KafkaCosmosChangeFeedState(projectedContinuation, feedRange, projectedItemLsn); + } + @Override public boolean equals(Object o) { if (this == o) { @@ -108,7 +121,11 @@ public KafkaCosmosChangeFeedState deserialize( final JsonNode rootNode = jsonParser.getCodec().readTree(jsonParser); String continuationState = rootNode.get("responseContinuation").asText(); FeedRange targetRange = FeedRange.fromString(rootNode.get("targetRange").asText()); - String continuationLsn = rootNode.get("itemLsn").asText(); + String continuationLsn = null; + if (rootNode.hasNonNull("itemLsn")) { + String itemLsnValue = rootNode.get("itemLsn").asText(); + continuationLsn = "null".equals(itemLsnValue) ? null : itemLsnValue; + } return new KafkaCosmosChangeFeedState(continuationState, targetRange, continuationLsn); } } diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSourceConnectorTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSourceConnectorTest.java index 398fafa1f2d9..09570c0a4da7 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSourceConnectorTest.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/CosmosSourceConnectorTest.java @@ -372,6 +372,19 @@ public void getTaskConfigsAfterSplit() throws JsonProcessingException { multiPartitionContainer.getResourceId(), FeedRange.forFullRange()); + List currentFeedRanges = cosmosAsyncClient + .getDatabase(databaseName) + .getContainer(multiPartitionContainer.getId()) + .getFeedRanges() + .block(); + List childContinuationTokens = new ArrayList<>(); + for (int i = 0; i < currentFeedRanges.size(); i++) { + childContinuationTokens.add( + new CompositeContinuationToken( + String.valueOf(100 + i), + ((FeedRangeEpkImpl) currentFeedRanges.get(i)).getRange())); + } + String initialContinuationState = new ChangeFeedStateV1( multiPartitionContainer.getResourceId(), FeedRangeEpkImpl.forFullRange(), @@ -380,10 +393,10 @@ public void getTaskConfigsAfterSplit() throws JsonProcessingException { FeedRangeContinuation.create( multiPartitionContainer.getResourceId(), FeedRangeEpkImpl.forFullRange(), - Arrays.asList(new CompositeContinuationToken("1", FeedRangeEpkImpl.forFullRange().getRange())))).toString(); + childContinuationTokens)).toString(); FeedRangeContinuationTopicOffset feedRangeContinuationTopicOffset = - new FeedRangeContinuationTopicOffset(initialContinuationState, "1"); // using the same itemLsn as in the continuationToken + new FeedRangeContinuationTopicOffset(initialContinuationState, "100"); Map, Map> initialOffsetMap = new HashMap<>(); initialOffsetMap.put( FeedRangeContinuationTopicPartition.toMap(feedRangeContinuationTopicPartition), @@ -844,11 +857,13 @@ private List getFeedRangeTaskUnits( KafkaCosmosChangeFeedState kafkaCosmosChangeFeedState = null; if (StringUtils.isNotEmpty(continuationState)) { ChangeFeedState changeFeedState = ChangeFeedStateV1.fromString(continuationState); + ChangeFeedState projectedState = changeFeedState.extractForEffectiveRange( + ((FeedRangeEpkImpl) feedRange).getRange()); kafkaCosmosChangeFeedState = new KafkaCosmosChangeFeedState( - continuationState, + projectedState.toString(), feedRange, - changeFeedState.getContinuation().getCurrentContinuationToken().getToken()); + null); } return new FeedRangeTaskUnit( diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTaskTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTaskTest.java index 1e9357ff0f50..47f2ba87d523 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTaskTest.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/CosmosSourceTaskTest.java @@ -5,16 +5,29 @@ import com.azure.cosmos.CosmosAsyncClient; import com.azure.cosmos.CosmosAsyncContainer; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedMode; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStartFromInternal; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedState; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStateV1; +import com.azure.cosmos.implementation.feedranges.FeedRangeContinuation; +import com.azure.cosmos.implementation.feedranges.FeedRangeEpkImpl; +import com.azure.cosmos.implementation.query.CompositeContinuationToken; +import com.azure.cosmos.implementation.routing.PartitionKeyInternalHelper; import com.azure.cosmos.kafka.connect.InMemoryStorageReader; import com.azure.cosmos.kafka.connect.KafkaCosmosTestConfigurations; import com.azure.cosmos.kafka.connect.KafkaCosmosTestSuiteBase; import com.azure.cosmos.kafka.connect.TestItem; import com.azure.cosmos.kafka.connect.implementation.CosmosClientCache; import com.azure.cosmos.kafka.connect.implementation.CosmosClientCacheItem; +import com.azure.cosmos.models.CosmosChangeFeedRequestOptions; import com.azure.cosmos.models.CosmosContainerProperties; import com.azure.cosmos.models.CosmosItemRequestOptions; import com.azure.cosmos.models.CosmosQueryRequestOptions; import com.azure.cosmos.models.FeedRange; +import com.azure.cosmos.models.FeedResponse; +import com.azure.cosmos.models.ModelBridgeInternal; +import com.azure.cosmos.models.PartitionKey; +import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.models.ThroughputProperties; import com.azure.cosmos.models.ThroughputResponse; import org.apache.kafka.connect.data.Struct; @@ -26,6 +39,7 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -226,6 +240,99 @@ public void poll_splitWhenStartFeedRangeTask() { } } } + + @Test(groups = { "kafka" }, timeOut = 60 * TIMEOUT) + public void pollWithDivergentChildContinuations() { + Map sourceConfigMap = new HashMap<>(); + sourceConfigMap.put("azure.cosmos.account.endpoint", KafkaCosmosTestConfigurations.HOST); + sourceConfigMap.put("azure.cosmos.account.key", KafkaCosmosTestConfigurations.MASTER_KEY); + sourceConfigMap.put("azure.cosmos.source.database.name", databaseName); + sourceConfigMap.put( + "azure.cosmos.source.containers.includedList", + Arrays.asList(multiPartitionContainerName).toString()); + sourceConfigMap.put("azure.cosmos.source.task.id", UUID.randomUUID().toString()); + + CosmosSourceConfig sourceConfig = new CosmosSourceConfig(sourceConfigMap); + CosmosClientCacheItem clientItem = + CosmosClientCache.getCosmosClient(sourceConfig.getAccountConfig(), "pollWithDivergentChildContinuations"); + CosmosSourceTask sourceTask = null; + + try { + CosmosAsyncContainer container = clientItem + .getClient() + .getDatabase(databaseName) + .getContainer(multiPartitionContainerName); + CosmosContainerProperties containerProperties = container.read().block().getProperties(); + List childRanges = container.getFeedRanges().block().stream() + .map(range -> (FeedRangeEpkImpl) range) + .sorted(Comparator.comparing(range -> range.getRange().getMin())) + .collect(Collectors.toList()); + assertThat(childRanges.size()).isGreaterThanOrEqualTo(2); + + FeedRangeEpkImpl childA = childRanges.get(0); + FeedRangeEpkImpl childB = childRanges.get(1); + ChangeFeedState baselineB = snapshotNow(container, childB); + + ChangeFeedState advancedA = snapshotNow(container, childA); + int writesToChildA = 0; + while (numericToken(advancedA) <= numericToken(baselineB) && writesToChildA < 2_000) { + for (int i = 0; i < 50; i++) { + container.createItem(new TestItem( + UUID.randomUUID().toString(), + partitionKeyForRange("advance-a", childA, containerProperties.getPartitionKeyDefinition()), + "advance-a")).block(); + writesToChildA++; + } + advancedA = snapshotNow(container, childA); + } + assertThat(numericToken(advancedA)).isGreaterThan(numericToken(baselineB)); + + ChangeFeedState parentState = new ChangeFeedStateV1( + containerProperties.getResourceId(), + FeedRangeEpkImpl.forFullRange(), + ChangeFeedMode.INCREMENTAL, + ChangeFeedStartFromInternal.createFromNow(), + FeedRangeContinuation.create( + containerProperties.getResourceId(), + FeedRangeEpkImpl.forFullRange(), + Arrays.asList( + new CompositeContinuationToken(rawToken(advancedA), childA.getRange()), + new CompositeContinuationToken(rawToken(baselineB), childB.getRange())))); + + String markerId = UUID.randomUUID().toString(); + container.createItem( + new TestItem( + markerId, + partitionKeyForRange("marker-b", childB, containerProperties.getPartitionKeyDefinition()), + "marker-b")).block(); + + FeedRangeTaskUnit feedRangeTaskUnit = new FeedRangeTaskUnit( + databaseName, + multiPartitionContainerName, + containerProperties.getResourceId(), + childB, + new KafkaCosmosChangeFeedState(parentState.toString(), childB), + multiPartitionContainerName); + Map taskConfigMap = sourceConfig.originalsStrings(); + taskConfigMap.putAll( + CosmosSourceTaskConfig.getFeedRangeTaskUnitsConfigMap(Arrays.asList(feedRangeTaskUnit))); + + sourceTask = new CosmosSourceTask(); + sourceTask.initialize(new TestSourceTaskContext(taskConfigMap)); + sourceTask.start(taskConfigMap); + + List records = sourceTask.poll(); + assertThat(records.stream().anyMatch(record -> + markerId.equals(((Struct) record.value()).get("id").toString()))).isTrue(); + } finally { + if (sourceTask != null) { + sourceTask.stop(); + } + CosmosClientCache.releaseCosmosClient(clientItem.getClientConfig()); + clientItem.getClient().close(); + } + } + @Test(groups = { "kafka", "kafka-emulator" }, timeOut = TIMEOUT) public void pollWithSpecificFeedRange() { // Test only items belong to the feedRange defined in the feedRangeTaskUnit will be returned @@ -513,6 +620,43 @@ private List createItems( return testItems; } + private ChangeFeedState snapshotNow(CosmosAsyncContainer container, FeedRangeEpkImpl feedRange) { + FeedResponse response = container + .queryChangeFeed( + CosmosChangeFeedRequestOptions.createForProcessingFromNow(feedRange), + com.fasterxml.jackson.databind.JsonNode.class) + .byPage() + .next() + .block(); + return ChangeFeedState.fromString(response.getContinuationToken()); + } + + private String partitionKeyForRange( + String prefix, + FeedRangeEpkImpl targetRange, + PartitionKeyDefinition partitionKeyDefinition) { + + for (int i = 0; i < 100_000; i++) { + String id = prefix + "-" + UUID.randomUUID(); + String effectivePartitionKey = PartitionKeyInternalHelper.getEffectivePartitionKeyString( + ModelBridgeInternal.getPartitionKeyInternal(new PartitionKey(id)), + partitionKeyDefinition); + if (targetRange.getRange().contains(effectivePartitionKey)) { + return id; + } + } + + throw new IllegalStateException("Unable to generate an id for feed range " + targetRange); + } + + private long numericToken(ChangeFeedState state) { + return Long.parseLong(rawToken(state).replace("\"", "")); + } + + private String rawToken(ChangeFeedState state) { + return state.getContinuation().getCurrentContinuationToken().getToken(); + } + public static class TestSourceTaskContext implements SourceTaskContext { private final Map map; private final OffsetStorageReader offsetStorageReader; diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedStateTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedStateTest.java new file mode 100644 index 000000000000..b932e96b795a --- /dev/null +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/source/KafkaCosmosChangeFeedStateTest.java @@ -0,0 +1,82 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.kafka.connect.implementation.source; + +import com.azure.cosmos.implementation.Utils; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedMode; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStartFromInternal; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedState; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStateV1; +import com.azure.cosmos.implementation.feedranges.FeedRangeContinuation; +import com.azure.cosmos.implementation.feedranges.FeedRangeEpkImpl; +import com.azure.cosmos.implementation.query.CompositeContinuationToken; +import com.azure.cosmos.implementation.routing.Range; +import org.testng.annotations.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public class KafkaCosmosChangeFeedStateTest { + private static final Range LOW_RANGE = new Range<>("", "80", true, false); + private static final Range HIGH_RANGE = new Range<>("80", "FF", true, false); + private static final FeedRangeEpkImpl FULL_RANGE = + new FeedRangeEpkImpl(new Range<>("", "FF", true, false)); + + @Test(groups = "unit") + public void extractForFeedRangePreservesChildContinuationAndDropsParentItemLsn() { + KafkaCosmosChangeFeedState parentState = createParentState("2050"); + + KafkaCosmosChangeFeedState highChild = + parentState.extractForFeedRange(new FeedRangeEpkImpl(HIGH_RANGE)); + List childTokens = + ChangeFeedState.fromString(highChild.getResponseContinuation()).extractContinuationTokens(); + + assertThat(childTokens).hasSize(1); + assertThat(childTokens.get(0).getRange()).isEqualTo(HIGH_RANGE); + assertThat(childTokens.get(0).getToken()).isEqualTo("1000"); + assertThat(highChild.getItemLsn()).isNull(); + } + + @Test(groups = "unit") + public void extractForExactFeedRangePreservesItemLsn() { + KafkaCosmosChangeFeedState parentState = createParentState("2050"); + + KafkaCosmosChangeFeedState exactState = parentState.extractForFeedRange(FULL_RANGE); + + assertThat(exactState.getItemLsn()).isEqualTo("2050"); + assertThat(ChangeFeedState.fromString(exactState.getResponseContinuation()).extractContinuationTokens()) + .hasSize(2); + } + + @Test(groups = "unit") + public void serializationPreservesNullItemLsn() throws Exception { + KafkaCosmosChangeFeedState childState = createParentState(null) + .extractForFeedRange(new FeedRangeEpkImpl(HIGH_RANGE)); + + String json = Utils.getSimpleObjectMapper().writeValueAsString(childState); + KafkaCosmosChangeFeedState deserialized = + Utils.getSimpleObjectMapper().readValue(json, KafkaCosmosChangeFeedState.class); + + assertThat(deserialized.getItemLsn()).isNull(); + assertThat(deserialized).isEqualTo(childState); + } + + private static KafkaCosmosChangeFeedState createParentState(String itemLsn) { + ChangeFeedState parentState = new ChangeFeedStateV1( + "containerRid", + FULL_RANGE, + ChangeFeedMode.INCREMENTAL, + ChangeFeedStartFromInternal.createFromBeginning(), + FeedRangeContinuation.create( + "containerRid", + FULL_RANGE, + Arrays.asList( + new CompositeContinuationToken("2050", LOW_RANGE), + new CompositeContinuationToken("1000", HIGH_RANGE)))); + + return new KafkaCosmosChangeFeedState(parentState.toString(), FULL_RANGE, itemLsn); + } +} diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/KafkaChangeFeedContinuationProjectionTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/KafkaChangeFeedContinuationProjectionTest.java new file mode 100644 index 000000000000..04604370157d --- /dev/null +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/KafkaChangeFeedContinuationProjectionTest.java @@ -0,0 +1,145 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.implementation; + +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedMode; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStartFromInternal; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedState; +import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStateV1; +import com.azure.cosmos.implementation.feedranges.FeedRangeContinuation; +import com.azure.cosmos.implementation.feedranges.FeedRangeEpkImpl; +import com.azure.cosmos.implementation.query.CompositeContinuationToken; +import com.azure.cosmos.implementation.routing.Range; +import com.azure.cosmos.models.CosmosChangeFeedRequestOptions; +import com.azure.cosmos.models.FeedRange; +import com.azure.cosmos.models.ModelBridgeInternal; +import org.testng.annotations.Test; + +import java.util.Arrays; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +public class KafkaChangeFeedContinuationProjectionTest { + private static final String CONTAINER_RID = "containerRid"; + private static final Range LOW_RANGE = new Range<>("", "80", true, false); + private static final Range HIGH_RANGE = new Range<>("80", "FF", true, false); + private static final FeedRangeEpkImpl FULL_RANGE = + new FeedRangeEpkImpl(new Range<>("", "FF", true, false)); + + @Test(groups = "unit") + public void nullItemLsnPreservesChildContinuations() { + ChangeFeedState parentState = createParentState("2050", "1000"); + + assertProjectedToken(parentState, new FeedRangeEpkImpl(LOW_RANGE), null, LOW_RANGE, "2050"); + assertProjectedToken(parentState, new FeedRangeEpkImpl(HIGH_RANGE), null, HIGH_RANGE, "1000"); + } + + @Test(groups = "unit") + public void itemLsnOnlyOverridesOwningChildContinuation() { + ChangeFeedState parentState = createParentState("2050", "1000"); + + assertProjectedToken(parentState, new FeedRangeEpkImpl(LOW_RANGE), "2055", LOW_RANGE, "2055"); + assertProjectedToken(parentState, new FeedRangeEpkImpl(HIGH_RANGE), "2055", HIGH_RANGE, "1000"); + } + + @Test(groups = "unit") + public void itemLsnOwnershipFollowsCurrentTokenRange() { + ChangeFeedState parentState = createParentState( + Arrays.asList( + new CompositeContinuationToken("1000", HIGH_RANGE), + new CompositeContinuationToken("2050", LOW_RANGE))); + + assertProjectedToken(parentState, new FeedRangeEpkImpl(LOW_RANGE), "1005", LOW_RANGE, "2050"); + assertProjectedToken(parentState, new FeedRangeEpkImpl(HIGH_RANGE), "1005", HIGH_RANGE, "1005"); + } + + @Test(groups = "unit") + public void itemLsnOnlyOverridesCurrentTokenForFullRangeTarget() { + ChangeFeedState parentState = createParentState( + Arrays.asList( + new CompositeContinuationToken("1000", HIGH_RANGE), + new CompositeContinuationToken("2050", LOW_RANGE))); + + CosmosChangeFeedRequestOptions options = + ImplementationBridgeHelpers.CosmosChangeFeedRequestOptionsHelper + .getCosmosChangeFeedRequestOptionsAccessor() + .createForProcessingFromContinuation(parentState.toString(), FULL_RANGE, "1005"); + List projectedTokens = + ModelBridgeInternal.getChangeFeedContinuationState(options).extractContinuationTokens(); + + assertThat(projectedTokens).hasSize(2); + assertThat(projectedTokens.get(0).getRange()).isEqualTo(LOW_RANGE); + assertThat(projectedTokens.get(0).getToken()).isEqualTo("2050"); + assertThat(projectedTokens.get(1).getRange()).isEqualTo(HIGH_RANGE); + assertThat(projectedTokens.get(1).getToken()).isEqualTo("1005"); + } + + @Test(groups = "unit") + public void extractedContinuationIsScopedToTargetChild() { + ChangeFeedState parentState = createParentState("2050", "1000"); + + String projectedContinuation = ImplementationBridgeHelpers.CosmosChangeFeedRequestOptionsHelper + .getCosmosChangeFeedRequestOptionsAccessor() + .extractContinuationForFeedRange(parentState.toString(), new FeedRangeEpkImpl(HIGH_RANGE)); + + List projectedTokens = + ChangeFeedState.fromString(projectedContinuation).extractContinuationTokens(); + assertThat(projectedTokens).hasSize(1); + assertThat(projectedTokens.get(0).getRange()).isEqualTo(HIGH_RANGE); + assertThat(projectedTokens.get(0).getToken()).isEqualTo("1000"); + } + + @Test(groups = "unit") + public void nonOverlappingTargetFails() { + ChangeFeedState parentState = createParentState("2050", "1000"); + FeedRange nonOverlappingRange = + new FeedRangeEpkImpl(new Range<>("FF", "FFFF", true, false)); + + assertThatThrownBy(() -> ImplementationBridgeHelpers.CosmosChangeFeedRequestOptionsHelper + .getCosmosChangeFeedRequestOptionsAccessor() + .createForProcessingFromContinuation(parentState.toString(), nonOverlappingRange, null)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("does not overlap"); + } + + private static void assertProjectedToken( + ChangeFeedState parentState, + FeedRange targetRange, + String itemLsn, + Range expectedRange, + String expectedToken) { + + CosmosChangeFeedRequestOptions options = + ImplementationBridgeHelpers.CosmosChangeFeedRequestOptionsHelper + .getCosmosChangeFeedRequestOptionsAccessor() + .createForProcessingFromContinuation(parentState.toString(), targetRange, itemLsn); + List projectedTokens = + ModelBridgeInternal.getChangeFeedContinuationState(options).extractContinuationTokens(); + + assertThat(projectedTokens).hasSize(1); + assertThat(projectedTokens.get(0).getRange()).isEqualTo(expectedRange); + assertThat(projectedTokens.get(0).getToken()).isEqualTo(expectedToken); + } + + private static ChangeFeedState createParentState(String lowToken, String highToken) { + return createParentState( + Arrays.asList( + new CompositeContinuationToken(lowToken, LOW_RANGE), + new CompositeContinuationToken(highToken, HIGH_RANGE))); + } + + private static ChangeFeedState createParentState(List tokens) { + return new ChangeFeedStateV1( + CONTAINER_RID, + FULL_RANGE, + ChangeFeedMode.INCREMENTAL, + ChangeFeedStartFromInternal.createFromBeginning(), + FeedRangeContinuation.create( + CONTAINER_RID, + FULL_RANGE, + tokens)); + } +} diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index 9cbcee519b89..c6ab0985ab7e 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -12,6 +12,7 @@ * Unified request-level consistency override behavior across transports: invalid attempts to upgrade the request consistency level above the account default are now silently ignored instead of returning `BadRequest` in some gateway paths. - See PR [49606](https://github.com/Azure/azure-sdk-for-java/pull/49606). * Fixed `partitionLevelCircuitBreakerCfg` missing from the `clientCfgs` section of `CosmosDiagnostics` when Per-Partition Circuit Breaker is explicitly enabled. - See PR [49734](https://github.com/Azure/azure-sdk-for-java/pull/49734). * Fixed thin-client (Gateway V2) queries with a prefix (partial) hierarchical partition key returning co-located documents from other logical partitions. - See PR [49688](https://github.com/Azure/azure-sdk-for-java/pull/49688). +* Fixed Kafka change feed continuation projection after partition splits so each child range resumes from its own continuation instead of a sibling range's LSN. - See [PR 50031](https://github.com/Azure/azure-sdk-for-java/pull/50031) #### Other Changes * Reduced memory footprint of deserialized `PartitionKeyRange` instances by stripping unused fields in the `PartitionKeyRange(ObjectNode)` constructor - See PR [49513](https://github.com/Azure/azure-sdk-for-java/pull/49513). diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java index 92ef82a65699..517e73ac77bb 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java @@ -433,6 +433,7 @@ public interface CosmosChangeFeedRequestOptionsAccessor { OperationContextAndListenerTuple getOperationContext(CosmosChangeFeedRequestOptions changeFeedRequestOptions); CosmosDiagnosticsThresholds getDiagnosticsThresholds(CosmosChangeFeedRequestOptions options); CosmosChangeFeedRequestOptions createForProcessingFromContinuation(String continuation, FeedRange targetRange, String continuationLsn); + String extractContinuationForFeedRange(String continuation, FeedRange targetRange); CosmosChangeFeedRequestOptions clone(CosmosChangeFeedRequestOptions toBeCloned); diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/models/CosmosChangeFeedRequestOptions.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/models/CosmosChangeFeedRequestOptions.java index 22b5e6242d97..5b4116bb4e64 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/models/CosmosChangeFeedRequestOptions.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/models/CosmosChangeFeedRequestOptions.java @@ -25,7 +25,7 @@ import com.azure.cosmos.util.Beta; import java.time.Instant; -import java.util.Arrays; +import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -318,33 +318,68 @@ public static CosmosChangeFeedRequestOptions createForProcessingFromContinuation */ static CosmosChangeFeedRequestOptions createForProcessingFromContinuation( String continuation, FeedRange targetRange, String continuationLsn) { - if (targetRange instanceof FeedRangeEpkImpl) { - Range normalizedRange = - FeedRangeInternal.normalizeRange(((FeedRangeEpkImpl) targetRange).getRange()); + return createForProcessingFromContinuation( + projectContinuationToFeedRange(continuation, targetRange, continuationLsn)); + } + + static String extractContinuationForFeedRange(String continuation, FeedRange targetRange) { + return projectContinuationToFeedRange(continuation, targetRange, null).toString(); + } - final ChangeFeedState changeFeedState = ChangeFeedState.fromString(continuation); + private static ChangeFeedState projectContinuationToFeedRange( + String continuation, + FeedRange targetRange, + String continuationLsn) { - if (StringUtils.isEmpty(continuationLsn)) { - continuationLsn = changeFeedState.getContinuation().getCurrentContinuationToken().getToken(); - } + if (!(targetRange instanceof FeedRangeEpkImpl)) { + throw new IllegalStateException( + "createForProcessingFromContinuation does not support feedRange type " + + (targetRange == null ? "null" : targetRange.getClass())); + } + + Range normalizedRange = + FeedRangeInternal.normalizeRange(((FeedRangeEpkImpl) targetRange).getRange()); + ChangeFeedState parentState = ChangeFeedState.fromString(continuation); + FeedRangeContinuation parentContinuation = parentState.getContinuation(); + if (parentContinuation == null) { + throw new IllegalStateException("The change feed continuation does not contain continuation tokens."); + } - ChangeFeedState targetChangeFeedState = - new ChangeFeedStateV1( - changeFeedState.getContainerRid(), - (FeedRangeEpkImpl) targetRange, - changeFeedState.getMode(), - changeFeedState.getStartFromSettings(), - FeedRangeContinuation.create( - changeFeedState.getContainerRid(), - (FeedRangeEpkImpl) targetRange, - Arrays.asList(new CompositeContinuationToken(continuationLsn, normalizedRange)) - ) - ); - - return createForProcessingFromContinuation(targetChangeFeedState); + CompositeContinuationToken parentCurrentToken = parentContinuation.getCurrentContinuationToken(); + boolean overlapsTarget = parentState.extractContinuationTokens().stream() + .anyMatch(token -> Range.checkOverlapping(token.getRange(), normalizedRange)); + if (!overlapsTarget) { + throw new IllegalStateException( + "The change feed continuation does not overlap target feed range " + targetRange); } - throw new IllegalStateException("createForProcessingFromContinuation does not support feedRange type " + targetRange.getClass()); + ChangeFeedState targetState = parentState.extractForEffectiveRange(normalizedRange); + List targetTokens = targetState.extractContinuationTokens(); + + if (StringUtils.isEmpty(continuationLsn) + || parentCurrentToken == null + || !Range.checkOverlapping(parentCurrentToken.getRange(), normalizedRange)) { + return targetState; + } + + List updatedTokens = new ArrayList<>(targetTokens.size()); + for (CompositeContinuationToken targetToken : targetTokens) { + String token = Range.checkOverlapping(parentCurrentToken.getRange(), targetToken.getRange()) + ? continuationLsn + : targetToken.getToken(); + updatedTokens.add(new CompositeContinuationToken(token, targetToken.getRange())); + } + + FeedRangeEpkImpl effectiveTargetRange = (FeedRangeEpkImpl) targetState.getFeedRange(); + return new ChangeFeedStateV1( + targetState.getContainerRid(), + effectiveTargetRange, + targetState.getMode(), + targetState.getStartFromSettings(), + FeedRangeContinuation.create( + targetState.getContainerRid(), + effectiveTargetRange, + updatedTokens)); } static CosmosChangeFeedRequestOptions createForProcessingFromContinuation( @@ -752,6 +787,11 @@ public CosmosChangeFeedRequestOptions createForProcessingFromContinuation( return CosmosChangeFeedRequestOptions.createForProcessingFromContinuation(continuation, targetRange, continuationLsn); } + @Override + public String extractContinuationForFeedRange(String continuation, FeedRange targetRange) { + return CosmosChangeFeedRequestOptions.extractContinuationForFeedRange(continuation, targetRange); + } + @Override public CosmosChangeFeedRequestOptions clone(CosmosChangeFeedRequestOptions toBeCloned) { return new CosmosChangeFeedRequestOptions(toBeCloned);