Skip to content

Fix Kafka patch filter precondition handling #50053

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

Open
Arnab Nandy (arnabnandy7) wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix Kafka patch filter precondition handling #50053

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion sdk/cosmos/azure-cosmos-kafka-connect/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)

2 changes: 1 addition & 1 deletion sdk/cosmos/azure-cosmos-kafka-connect/docs/configuration-reference.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. |
6 changes: 5 additions & 1 deletion ...ct/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosBulkWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -373,4 +378,3 @@ public boolean onEmitFailure(SignalType signalType, Sinks.EmitResult emitResult)
}
}
}

6 changes: 4 additions & 2 deletions ...t/src/main/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosPointWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
);
}
Expand Down Expand Up @@ -242,4 +245,3 @@ private CosmosItemRequestOptions getCosmosItemRequestOptions() {
return itemRequestOptions;
}
}

8 changes: 8 additions & 0 deletions .../src/test/java/com/azure/cosmos/kafka/connect/implementation/sink/CosmosSinkTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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<ObjectNode> itemsAfterReplay = this.getAllItems(container);
assertThat(itemsAfterReplay.size()).isEqualTo(itemsFromContainer.size());
assertThat(itemsAfterReplay.containsAll(itemsFromContainer)).isTrue();
} finally {
if (cosmosClient != null) {
cleanUpContainer(cosmosClient, databaseName, singlePartitionContainerProperties.getId());
Expand Down