diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md index 65d99b6cf6fa..96b17cd41927 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md @@ -8,6 +8,8 @@ #### Bugs Fixed +* Fixed `ItemPatch` writes to skip items excluded by `azure.cosmos.sink.write.patch.filter` instead of failing with `412 Precondition Failed`. + #### Other Changes ### 2.11.0 (2026-06-08) @@ -132,4 +134,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/docs/configuration-reference.md b/sdk/cosmos/azure-cosmos-kafka-connect/docs/configuration-reference.md index ecd2cbbf660a..c848fe0432fc 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/docs/configuration-reference.md +++ b/sdk/cosmos/azure-cosmos-kafka-connect/docs/configuration-reference.md @@ -64,4 +64,4 @@ | `azure.cosmos.sink.id.strategy` | `ProvidedInValueStrategy` | A strategy used to populate the document with an ``id``. Valid strategies are: ``TemplateStrategy``, ``FullKeyStrategy``, ``KafkaMetadataStrategy``, ``ProvidedInKeyStrategy``, ``ProvidedInValueStrategy``. Configuration properties prefixed with``id.strategy`` are passed through to the strategy. For example, when using ``id.strategy=TemplateStrategy`` , the property ``id.strategy.template`` is passed through to the template strategy and used to specify the template string to be used in constructing the ``id``. | | `azure.cosmos.sink.write.patch.operationType.default` | `Set` | Default Cosmos DB patch operation type. Supported ones include none, add, set, replace, remove, increment. Choose none for no-op, for others please reference [here](https://docs.microsoft.com/azure/cosmos-db/partial-document-update#supported-operations) for full context. | | `azure.cosmos.sink.write.patch.property.configs` | `""` | Cosmos DB patch json property configs. It can contain multiple definitions matching the following patterns separated by comma. property(jsonProperty).op(operationType) or property(jsonProperty).path(patchInCosmosdb).op(operationType) - The difference of the second pattern is that it also allows you to define a different cosmosdb path. Note: It does not support nested json property config. | -| `azure.cosmos.sink.write.patch.filter` | `""` | Used for [Conditional patch](https://docs.microsoft.com/azure/cosmos-db/partial-document-update-getting-started#java) | +| `azure.cosmos.sink.write.patch.filter` | `""` | Used for [Conditional patch](https://docs.microsoft.com/azure/cosmos-db/partial-document-update-getting-started#java). When the filter condition is not met, the item is left unchanged and the write is treated as a successful no-op. | diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosBulkWriter.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosBulkWriter.java index c736c2385d07..483fa63fd8d5 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosBulkWriter.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosBulkWriter.java @@ -321,6 +321,11 @@ private boolean shouldIgnore(BulkOperationFailedException failedException) { switch (this.writeConfig.getItemWriteStrategy()) { case ITEM_APPEND: return KafkaCosmosExceptionsHelper.isResourceExistsException(failedException); + case ITEM_PATCH: + // A 412 on patch can only originate from the configured filter predicate because etag/If-Match + // is not wired for patch today. If etag support is added, these two causes must be distinguished. + return StringUtils.isNotEmpty(this.writeConfig.getCosmosPatchConfig().getFilter()) + && KafkaCosmosExceptionsHelper.isPreconditionFailedException(failedException); case ITEM_DELETE: return KafkaCosmosExceptionsHelper.isNotFoundException(failedException); case ITEM_DELETE_IF_NOT_MODIFIED: @@ -373,4 +378,3 @@ public boolean onEmitFailure(SignalType signalType, Sinks.EmitResult emitResult) } } } - diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosPointWriter.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosPointWriter.java index f5cd2e749d9d..b125d4fbee5e 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosPointWriter.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosPointWriter.java @@ -183,7 +183,10 @@ private void patchWithRetry(CosmosAsyncContainer container, SinkOperation sinkOp ObjectNode.class); }).then(); }, - (throwable) -> false, // no exceptions should be ignored + // A 412 on patch can only originate from the configured filter predicate because etag/If-Match + // is not wired for patch today. If etag support is added, these two causes must be distinguished. + (throwable) -> StringUtils.isNotEmpty(this.writeConfig.getCosmosPatchConfig().getFilter()) + && KafkaCosmosExceptionsHelper.isPreconditionFailedException(throwable), sinkOperation ); } @@ -242,4 +245,3 @@ private CosmosItemRequestOptions getCosmosItemRequestOptions() { return itemRequestOptions; } } - diff --git a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTaskTest.java b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTaskTest.java index 9d85408fb0f5..fb09e9f47478 100644 --- a/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTaskTest.java +++ b/sdk/cosmos/azure-cosmos-kafka-connect/src/test/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTaskTest.java @@ -534,6 +534,7 @@ public void sinkWithItemPatch(boolean bulkEnabled) { + " property(doubleProperty).op(add)," + " property(arrayProperty).path(/listProperty/0).op(replace)," + " property(toBeRemovedProperty).op(remove)"); + sinkConfigMap.put("azure.cosmos.sink.write.patch.filter", "FROM c WHERE c.intProperty = 1"); sinkConfigMap.put("azure.cosmos.sink.task.id", UUID.randomUUID().toString()); CosmosSinkTask sinkTask = new CosmosSinkTask(); @@ -607,6 +608,13 @@ public void sinkWithItemPatch(boolean bulkEnabled) { assertThat(expectedItem.get("doubleProperty").doubleValue()).isEqualTo(itemFromContainer.get("doubleProperty").doubleValue()); assertThat(expectedItem.get("toBeRemovedProperty")).isNull(); } + + // Replay the same records. The filter now excludes every item, so the resulting 412 responses + // should be treated as successful no-op skips by both the point and bulk writers. + sinkTask.put(sinkRecordList); + List itemsAfterReplay = this.getAllItems(container); + assertThat(itemsAfterReplay.size()).isEqualTo(itemsFromContainer.size()); + assertThat(itemsAfterReplay.containsAll(itemsFromContainer)).isTrue(); } finally { if (cosmosClient != null) { cleanUpContainer(cosmosClient, databaseName, singlePartitionContainerProperties.getId());