apache/seatunnel · warning · IOException

Failed to close Pulsar consumer.

Error message

Failed to close Pulsar consumer.

What it means

Thrown by PulsarSplitReaderThread.closeConsumer as an IOException when consumer.close() fails. Close is invoked from run() and close(), so it can surface on normal task completion or cancellation. It indicates the Pulsar consumer could not gracefully unsubscribe/close, typically because the client connection is broken.

Source

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

        try {
            return consumerBuilder.subscribe();
        } catch (PulsarClientException e) {
            throw new PulsarConnectorException(
                    PulsarConnectorErrorCode.OPEN_PULSAR_ADMIN_FAILED,
                    "Failed to create pulsar consumer:",
                    e);
        }
    }

    /**
     * Closes the Pulsar consumer while exposing the connector classloader to Pulsar cleanup code.
     */
    private void closeConsumer() throws IOException {
        if (consumer != null) {
            try {
                PulsarConfigUtil.runWithConnectorClassLoader(consumer::close);
            } catch (Exception e) {
                throw new IOException("Failed to close Pulsar consumer.", e);
            }
        }
    }
}

View on GitHub (pinned to cf67b549a7)

Solutions

1. Check the IOException cause: AlreadyClosed is harmless during shutdown and can be ignored/logged at WARN.
2. Ensure the PulsarClient is not closed before the consumer (close consumers first, then the client).
3. Increase close timeout / check broker availability if the close times out.
4. Avoid calling reader close concurrently with the reader thread shutdown path.
5. If persistent, upgrade to a Pulsar client version with idempotent close behavior.

Example fix

// before
thread.close(); // may throw IOException on stale consumer
// after
try {
    thread.close();
} catch (IOException e) {
    if (!(e.getCause() instanceof AlreadyClosedException)) {
        LOG.warn("Failed to close Pulsar consumer cleanly", e);
    }
}

try { thread.close(); } catch (IOException e) { if (!(e.getCause() instanceof PulsarClientException.AlreadyClosedException)) { LOG.warn("Consumer close failed", e); } }

When it happens

Trigger: run() finishing or close() tearing down while consumer != null and PulsarConfigUtil.runWithConnectorClassLoader(consumer::close) throws — e.g. PulsarClientException.AlreadyClosedException, client already destroyed, timeouts, or classloader errors.

Common situations: Broker went away mid-job so graceful close times out; reader thread already failed and consumer is stale; task cancellation racing with the reader loop's own closeConsumer; client closed earlier by another component.

Related errors

• Failed to close Google Pub/Sub subscriber
• Failed to close Pulsar client after aborting transactions.
• Failed to close Pulsar sink writer.
• Failed to close Pulsar admin.
• CommonErrorCodeDeprecated.READER_OPERATION_FAILED

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