{"record":{"id":"99b08f98e4bebd93","repo":"apache/seatunnel","slug":"pulsarconnectorerrorcode-ack-cumulate-failed","errorCode":"PulsarConnectorErrorCode.ACK_CUMULATE_FAILED","errorMessage":"pulsar consumer acknowledgeCumulative failed.","messagePattern":"pulsar consumer acknowledgeCumulative failed\\.","errorType":"error_code","errorClass":"PulsarConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java","lineNumber":270,"sourceCode":"        }\n        try {\n            PulsarSplitReaderThread pulsarSplitReaderThread = splitReaders.get(splitId);\n            pulsarSplitReaderThread.committingCursor(messageId);\n\n            if (pendingCursorsToFinish.containsKey(splitId)\n                    && pendingCursorsToFinish.get(splitId).compareTo(messageId) == 0) {\n                finishedSplits.add(splitId);\n                try {\n                    pulsarSplitReaderThread.close();\n                } catch (IOException e) {\n                    throw new PulsarConnectorException(\n                            CommonErrorCodeDeprecated.READER_OPERATION_FAILED,\n                            \"Failed to close the split reader thread.\",\n                            e);\n                }\n            }\n        } catch (PulsarClientException e) {\n            throw new PulsarConnectorException(\n                    PulsarConnectorErrorCode.ACK_CUMULATE_FAILED,\n                    \"pulsar consumer acknowledgeCumulative failed.\",\n                    e);\n        }\n    }\n\n    private TablePath resolveTablePath(String splitId) {\n        TablePath tablePath = splitIdToTablePath.get(splitId);\n        return tablePath != null ? tablePath : defaultTablePath;\n    }\n\n    /**\n     * Preserves all cleanup failures while still allowing the remaining Pulsar resources to close.\n     */\n    private Throwable appendSuppressed(Throwable existingFailure, Throwable newFailure) {\n        if (existingFailure == null) {\n            return newFailure;\n        }","sourceCodeStart":252,"sourceCodeEnd":288,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java#L252-L288","documentation":"PulsarSourceReader.committingCursor acknowledges consumed messages cumulatively on the Pulsar consumer after committing a cursor. If the acknowledgeCumulative call throws PulsarClientException, it is wrapped as ACK_CUMULATE_FAILED with this message. Acknowledgement failure means the broker will redeliver messages, potentially causing duplicates after restart.","triggerScenarios":"During checkpoint commit, consumer.acknowledgeCumulative(messageId) throws PulsarClientException — broker unreachable, consumer already closed, or the messageId being acknowledged is invalid/not tracked by the consumer.","commonSituations":"Network blips or broker restarts between read and commit; consumer thread closed before acknowledgment (ordering issue); acknowledging a cursor position older than the consumer's tracked range.","solutions":["Check broker connectivity and consumer state at commit time; look at the wrapped PulsarClientException cause.","Rely on checkpoint/restart semantics: after a failed acknowledge, messages are redelivered — enable idempotent sinks to tolerate duplicates.","Ensure the split reader thread is still alive/connected when committingCursor runs; fix any close-ordering races.","Verify subscription type supports cumulative acknowledgment (Exclusive/Failover/Shared with cumulative ack semantics)."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    consumer.acknowledgeCumulative(messageId);\n} catch (PulsarClientException e) {\n    // retry with backoff; on final failure rely on broker redelivery\n}","preventionTips":["Enable idempotent downstream writes to tolerate redelivered messages.","Keep the consumer alive and connected until after acknowledgment.","Monitor broker availability; acknowledgment failures cluster around broker restarts.","Use a subscription type compatible with cumulative acknowledgment."],"tags":["pulsar","acknowledgment","checkpoint","consumer"],"backgroundTag":"http-error-response","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}