Fix handling of failed Cosmos bulk responses

[Arnab Nandy (arnabnandy7)]

Description

Fixes #50148.

Cosmos bulk execution reports certain per-item failures through CosmosBulkOperationResponse.getException(), while getResponse() is null. Spring Data Cosmos previously dereferenced getResponse() first, masking the original failure with a NullPointerException.

This pull request:

• Propagates bulk operation exceptions before accessing the item response.
• Routes the original exception through the existing Spring Data Cosmos exception and diagnostics handling.
• Applies the fix to reactive inserts, synchronous inserts, and bulk deletes.
• Adds unit coverage for successful and failed bulk responses.
• Updates the azure-spring-data-cosmos CHANGELOG.

All SDK Contribution checklist:

• The pull request does not introduce [breaking changes]
• CHANGELOG is updated for new features, bug fixes or other significant changes.
• I have read the contribution guidelines.

General Guidelines and Best Practices

• Title of the pull request is clear and informative.
• There are a small number of commits, each of which have an informative message. This means that previously merged commits do not appear in the history of the PR. For more information on cleaning up the commits in your PR, see this page.

Testing Guidelines

• Pull request includes test coverage for the included changes.

[github-actions]

Thank you for your contribution Arnab Nandy (@arnabnandy7)! We will review the pull request and get back to you soon.

[azure-pipelines]

Azure Pipelines:
Successfully started running 4 pipeline(s).
31 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

Note

Copilot was unable to run its full agentic suite in this review.

Adds centralized handling for failed Cosmos bulk operation responses to prevent null pointer exceptions and ensure failures are surfaced as reactive errors.

Changes:

• Introduced CosmosBulkOperationResponseUtils.emitErrorForFailedBulkOperation(...) to convert per-item bulk failures into Mono.error(...).
• Applied the utility in insertAll and deleteEntities bulk execution flows (sync + reactive templates).
• Added a unit test for the new utility and documented the fix in the changelog.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtils.java	New helper to emit reactive error when a bulk operation response contains an exception.
sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/ReactiveCosmosTemplate.java	Uses the helper to fail fast on per-item bulk failures during bulk insert.
sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosTemplate.java	Uses the helper to fail fast on per-item bulk failures during bulk insert and delete.
sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtilsUnitTest.java	Adds unit coverage for helper behavior.
sdk/spring/azure-spring-data-cosmos/CHANGELOG.md	Records the NPE fix related to bulk failures with missing item response.

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (2)

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosTemplate.java:825

• The new .handle(CosmosBulkOperationResponseUtils::emitErrorForFailedBulkOperation) changes failure behavior for deleteEntities, but the added tests only cover insertAll. Please add a unit test that mocks executeBulkOperations to emit a CosmosBulkOperationResponse with getException() != null (and missing item response), and assert deleteEntities(...) surfaces a CosmosAccessException with the bulk exception as the cause.

                .handle(CosmosBulkOperationResponseUtils::emitErrorForFailedBulkOperation)
                .onErrorResume(throwable ->
                    CosmosExceptionUtils.exceptionHandler("Failed to delete item(s)", throwable,
                        this.responseDiagnosticsProcessor))

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtilsUnitTest.java:30

• This test name is misleading: the operator doesn't 'return' a successful response; it emits it downstream. Consider renaming to something like emitErrorForFailedBulkOperationEmitsResponseWhenSuccessful to match the Reactor handle semantics.

    public void emitErrorForFailedBulkOperationReturnsSuccessfulResponse() {

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (4)

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosTemplateBulkFailureUnitTest.java:37

• This test hard-codes CONTAINER_NAME using BasicItem.class.getSimpleName(), which can drift from the real container name if BasicItem mapping/annotations change. Prefer deriving the container name from the same source used by the code under test (e.g., from entityInformation / mapping metadata) so the test fails only on behavioral regressions, not naming strategy changes.

    private static final String CONTAINER_NAME = BasicItem.class.getSimpleName();

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosTemplateBulkFailureUnitTest.java:60

• This test hard-codes CONTAINER_NAME using BasicItem.class.getSimpleName(), which can drift from the real container name if BasicItem mapping/annotations change. Prefer deriving the container name from the same source used by the code under test (e.g., from entityInformation / mapping metadata) so the test fails only on behavioral regressions, not naming strategy changes.

        when(client.getDatabase(DATABASE_NAME)).thenReturn(database);
        when(database.getContainer(CONTAINER_NAME)).thenReturn(container);

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosTemplate.java:314

• Using .handle(... sink.error(...)) will fail-fast on the first failed CosmosBulkOperationResponse, which cancels the remaining responses and can reduce diagnostics (and can also prevent observing successes when bulk responses contain a mix of successes/failures). If the intended contract is to process all responses and then surface aggregated failure information, consider collecting failures and erroring after consuming the stream (or emitting a richer exception that retains per-item details). If fail-fast is intended, it would be helpful to document that behavior near this operator.

            .getContainer(containerName)
            .executeBulkOperations(Flux.fromIterable(cosmosItemOperations), cosmosBulkExecutionOptions)
            .publishOn(CosmosSchedulers.SPRING_DATA_COSMOS_PARALLEL)
            .handle(CosmosBulkOperationResponseUtils::emitErrorForFailedBulkOperation)

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtils.java:22

• This utility encodes an important behavioral assumption (a bulk response may have getException() set and an absent item response, and callers should treat that as a stream error to avoid downstream null dereferences). Adding a brief class/method-level Javadoc explaining the Cosmos SDK response shape being handled here and why .handle(...) is used will make it less likely to be removed/changed in a way that reintroduces the original failure mode.

    static <TContext> void emitErrorForFailedBulkOperation(
        CosmosBulkOperationResponse<TContext> response,
        SynchronousSink<CosmosBulkOperationResponse<TContext>> sink) {
        if (response.getException() != null) {
            sink.error(response.getException());
        } else {
            sink.next(response);
        }
    }

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtilsUnitTest.java:32

• Current tests cover (1) getException() != null and (2) pass-through when an item response exists, but they don’t cover the regression scenario described in the changelog: a failed bulk operation with a missing/empty item response. Add a test case that builds a CosmosBulkOperationResponse with a missing item response (and the corresponding failure signal — exception and/or non-success status depending on API) to ensure the utility fails fast for that specific shape.

    public void emitErrorForFailedBulkOperationEmitsResponseWhenSuccessful() {
        CosmosBulkOperationResponse<Object> response = ModelBridgeInternal.createCosmosBulkOperationResponse(
            null, mock(CosmosBulkItemResponse.class), null);

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (3)

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtilsUnitTest.java:59

• This test currently relies on Mockito’s default boolean return value for itemResponse.isSuccessStatusCode() (implicitly false). To make the test robust and intention-revealing, explicitly stub when(itemResponse.isSuccessStatusCode()).thenReturn(false);.

        CosmosBulkItemResponse itemResponse = mock(CosmosBulkItemResponse.class);
        when(itemResponse.getStatusCode()).thenReturn(500);
        CosmosBulkOperationResponse<Object> response = ModelBridgeInternal.createCosmosBulkOperationResponse(
            null, itemResponse, null);

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtilsUnitTest.java:29

• expectErrorMatches(throwable -> throwable == exception) works, but yields less-informative failure output. Consider using expectErrorSatisfies with an explicit identity assertion (e.g., assertSame / AssertJ isSameAs) so failures show clearer diagnostics while keeping the same behavioral contract.

        StepVerifier.create(Flux.just(response)
                .handle(CosmosBulkOperationResponseUtils::emitErrorForFailedBulkOperation))
            .expectErrorMatches(throwable -> throwable == exception)
            .verify();

sdk/spring/azure-spring-data-cosmos/CHANGELOG.md:11

• Changelog entries typically refer to Java exception types using their proper class name. Consider changing 'null pointer exception' to NullPointerException (optionally formatted as inline code) for clarity and consistency.

* Fixed a null pointer exception when a bulk operation fails without an item response ([50148](https://github.com/Azure/azure-sdk-for-java/issues/50148)).

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (3)

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosTemplate.java:316

• Using handle(...)->sink.error(...) will terminate the Flux on the first failed per-item bulk response, which cancels the upstream subscription and can drop remaining item responses (including additional failures and diagnostics). If the goal is to report all failed items, consider consuming all responses and aggregating failures (e.g., collect failures then error with a composite) rather than failing on the first one. If fail-fast is intended, consider documenting that bulk operations now abort processing on the first failing item response.

            .getContainer(containerName)
            .executeBulkOperations(Flux.fromIterable(cosmosItemOperations), cosmosBulkExecutionOptions)
            .publishOn(CosmosSchedulers.SPRING_DATA_COSMOS_PARALLEL)
            // Fail fast so an individual bulk operation failure is not silently skipped.
            .handle(CosmosBulkOperationResponseUtils::emitErrorForFailedBulkOperation)
            .onErrorResume(throwable ->

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosBulkOperationResponseUtils.java:38

• response.getResponse() is called multiple times across branches. Consider storing it in a local variable (e.g., CosmosBulkItemResponse itemResponse = response.getResponse();) to make the flow easier to read and to avoid repeating the dereference/virtual calls.

        if (response.getException() != null) {
            sink.error(response.getException());
        } else if (response.getResponse() == null) {
            sink.error(new IllegalStateException("Bulk operation completed without an item response or exception."));
        } else if (!response.getResponse().isSuccessStatusCode()) {
            sink.error(new IllegalStateException(
                "Bulk operation failed with status code " + response.getResponse().getStatusCode() + "."));
        } else {
            sink.next(response);

sdk/spring/azure-spring-data-cosmos/CHANGELOG.md:11

• The code change introduces a broader behavioral shift than just preventing an NPE: per-item bulk failures (including unsuccessful status codes and missing responses without exceptions) are now converted into stream errors (and ultimately CosmosAccessException) rather than being ignored/filtered downstream. Consider expanding this changelog entry to explicitly mention the new fail-fast error propagation semantics so consumers aren’t surprised by exceptions where they may previously have seen partial success.

* Fixed a `NullPointerException` when a bulk operation fails without an item response ([50148](https://github.com/Azure/azure-sdk-for-java/issues/50148)).

[Copilot]

Pull request overview

Copilot reviewed 6 out of 6 changed files in this pull request and generated no new comments.

Suppressed comments (4)

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/ReactiveCosmosTemplate.java:552

• The comment is potentially misleading: the handle(...).error(...) will terminate downstream consumption, but it may not actually abort/cancel already-submitted server-side bulk operations. Consider rewording to something like 'Terminate processing on the first failed item response; remaining responses will not be consumed' to avoid implying Cosmos-side cancellation.

                             // Abort on the first failed item response; remaining bulk responses are not processed.
                             .handle(CosmosBulkOperationResponseUtils::emitErrorForFailedBulkOperation)

sdk/spring/azure-spring-data-cosmos/src/main/java/com/azure/spring/data/cosmos/core/CosmosTemplate.java:314

• Same concern as in ReactiveCosmosTemplate: this terminates the Reactor sequence but does not necessarily abort in-flight/queued bulk operations on the Cosmos side. Reword to avoid implying server-side abort.

            // Abort on the first failed item response; remaining bulk responses are not processed.
            .handle(CosmosBulkOperationResponseUtils::emitErrorForFailedBulkOperation)

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosTemplateBulkFailureUnitTest.java:57

• These tests depend on ModelBridgeInternal, which is an internal Cosmos SDK test hook and can be brittle across SDK changes. A more stable approach is to mock CosmosBulkOperationResponse with Mockito and stub getException() / getResponse() to represent each scenario, avoiding reliance on internal bridge APIs.

        CosmosBulkOperationResponse<Object> failedResponse = ModelBridgeInternal.createCosmosBulkOperationResponse(
            null, bulkException, null);

sdk/spring/azure-spring-data-cosmos/src/test/java/com/azure/spring/data/cosmos/core/CosmosTemplateBulkFailureUnitTest.java:77

• The templates now also fail fast when the item response is missing without an exception and when the item response is unsuccessful (non-2xx). The utility has coverage for these paths, but there’s no template-level assertion here verifying those errors are wrapped/propagated as expected (e.g., CosmosAccessException via onErrorResume). Consider adding at least one template test for each of those newly-wired behaviors to ensure the integration is covered.

    @Test
    public void insertAllPropagatesBulkExceptionWhenResponseIsMissing() {
        assertThatThrownBy(() -> cosmosTemplate.insertAll(entityInformation, Collections.singleton(entity)))
            .isInstanceOf(CosmosAccessException.class)
            .hasCause(bulkException);
    }