{"record":{"id":"8d8196567608bf4a","repo":"apache/seatunnel","slug":"error-while-closing-pulsar-consumer","errorCode":null,"errorMessage":"Error while closing pulsar consumer","messagePattern":"Error while closing pulsar consumer","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java","lineNumber":111,"sourceCode":"                if (message != null) {\n                    handover.produce(new RecordWithSplitId(message, split.splitId()));\n                    if (stopCursor.shouldStop(message)) {\n                        sourceReader.handleNoMoreElements(split.splitId(), message.getMessageId());\n                        break;\n                    }\n                } else {\n                    Thread.sleep(pollInterval);\n                }\n            }\n        } catch (Throwable t) {\n            LOG.error(\"Pulsar Consumer receive data error\", t);\n            handover.reportError(t);\n        } finally {\n            // make sure the PulsarConsumer is closed\n            try {\n                closeConsumer();\n            } catch (Throwable t) {\n                LOG.warn(\"Error while closing pulsar consumer\", t);\n            } finally {\n                running = false;\n            }\n        }\n    }\n\n    @Override\n    public void close() throws IOException {\n        running = false;\n        closeConsumer();\n    }\n\n    public void committingCursor(MessageId offsetsToCommit) throws PulsarClientException {\n        if (consumer == null) {\n            consumer = createPulsarConsumer(split);\n        }\n        consumer.acknowledgeCumulative(offsetsToCommit);\n    }","sourceCodeStart":93,"sourceCodeEnd":129,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSplitReaderThread.java#L93-L129","documentation":"When the Pulsar split reader thread finishes (normally or after an error), it attempts to close the Pulsar consumer in a finally block. If that close itself throws, the thread logs this warning. It is a secondary cleanup failure — the original cause of the thread exiting is reported separately via handover.reportError.","triggerScenarios":"The split reader thread's run() loop exits (exception reported to handover or normal end) and closeConsumer() in the finally block throws any Throwable — e.g. Pulsar client already shut down, broker unreachable during close, or consumer already closed.","commonSituations":"Job/task cancellation racing with consumer close; Pulsar client connection lost at shutdown; an earlier reader exception left the consumer in a bad state that also fails on close.","solutions":["Inspect the logged Throwable cause ('t') for the underlying close failure (usually a broker connectivity or client-shutdown issue)","Check broker availability and network stability at task shutdown time","If it appears at every job stop, verify Pulsar client version and graceful shutdown ordering; update connector"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// preflight broker reachability\nif (!pulsarAdmin.serviceUrlReachable()) throw new IllegalStateException(\"Pulsar broker unreachable\");","typeGuard":null,"tryCatchPattern":"try { /* run job */ } catch (org.apache.seatunnel.api.table.type.SeaTunnelRuntimeException e) { /* inspect suppressed close warning cause for broker connectivity */ log.error(\"pulsar close failed: {}\", e.getCause()); }","preventionTips":["Ensure broker is reachable during task shutdown","Avoid cancelling tasks during Pulsar client reconnect storms","Keep Pulsar client/connector versions aligned","Monitor broker availability at stop-the-world moments"],"tags":["pulsar","shutdown","cleanup"],"backgroundTag":"connection-refused","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"}