apache/seatunnel · error · PulsarConnectorException

PulsarConnectorErrorCode.ACK_CUMULATE_FAILED

PulsarConnectorErrorCode.ACK_CUMULATE_FAILED

Error message

pulsar consumer acknowledgeCumulative failed.

What it means

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.

Source

Thrown at seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java:270

        }
        try {
            PulsarSplitReaderThread pulsarSplitReaderThread = splitReaders.get(splitId);
            pulsarSplitReaderThread.committingCursor(messageId);

            if (pendingCursorsToFinish.containsKey(splitId)
                    && pendingCursorsToFinish.get(splitId).compareTo(messageId) == 0) {
                finishedSplits.add(splitId);
                try {
                    pulsarSplitReaderThread.close();
                } catch (IOException e) {
                    throw new PulsarConnectorException(
                            CommonErrorCodeDeprecated.READER_OPERATION_FAILED,
                            "Failed to close the split reader thread.",
                            e);
                }
            }
        } catch (PulsarClientException e) {
            throw new PulsarConnectorException(
                    PulsarConnectorErrorCode.ACK_CUMULATE_FAILED,
                    "pulsar consumer acknowledgeCumulative failed.",
                    e);
        }
    }

    private TablePath resolveTablePath(String splitId) {
        TablePath tablePath = splitIdToTablePath.get(splitId);
        return tablePath != null ? tablePath : defaultTablePath;
    }

    /**
     * Preserves all cleanup failures while still allowing the remaining Pulsar resources to close.
     */
    private Throwable appendSuppressed(Throwable existingFailure, Throwable newFailure) {
        if (existingFailure == null) {
            return newFailure;
        }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Check broker connectivity and consumer state at commit time; look at the wrapped PulsarClientException cause.
2. Rely on checkpoint/restart semantics: after a failed acknowledge, messages are redelivered — enable idempotent sinks to tolerate duplicates.
3. Ensure the split reader thread is still alive/connected when committingCursor runs; fix any close-ordering races.
4. Verify subscription type supports cumulative acknowledgment (Exclusive/Failover/Shared with cumulative ack semantics).

try {
    consumer.acknowledgeCumulative(messageId);
} catch (PulsarClientException e) {
    // retry with backoff; on final failure rely on broker redelivery
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "API error: {status}" and "HTTP 401/403/404/429/5xx" errors: non-2xx HTTP responses explained — this error's family across 27 libraries.

Related errors

• CommonErrorCodeDeprecated.READER_OPERATION_FAILED
• PulsarConnectorErrorCode.OPEN_PULSAR_ADMIN_FAILED
• Failed to close Pulsar consumer.
• ACKNOWLEDGE_FAILED
• COMMIT_FAILED

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/99b08f98e4bebd93. Report an issue: GitHub.