{"record":{"id":"f730138ae956497a","repo":"apache/seatunnel","slug":"commonerrorcodedeprecated-reader-operation-failed","errorCode":"CommonErrorCodeDeprecated.READER_OPERATION_FAILED","errorMessage":"Failed to start the split reader thread.","messagePattern":"Failed to start the split reader thread\\.","errorType":"error_code","errorClass":"PulsarConnectorException","httpStatus":null,"severity":"critical","filePath":"seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/reader/PulsarSourceReader.java","lineNumber":199,"sourceCode":"\n    @Override\n    public void addSplits(List<PulsarPartitionSplit> splits) {\n        for (PulsarPartitionSplit split : splits) {\n            splitStates.put(split.splitId(), split);\n            TablePath tablePath = resolveTablePath(split);\n            if (tablePath != null) {\n                splitIdToTablePath.put(split.splitId(), tablePath);\n            }\n            PulsarSplitReaderThread splitReaderThread = createPulsarSplitReaderThread(split);\n            try {\n                splitReaderThread.setName(\n                        \"Pulsar Source Data Consumer \" + split.getPartition().getPartition());\n                splitReaderThread.open();\n                splitReaders.put(split.splitId(), splitReaderThread);\n                splitReaderThread.start();\n                LOG.info(\"PulsarSplitReaderThread = {} start\", splitReaderThread.getName());\n            } catch (PulsarClientException e) {\n                throw new PulsarConnectorException(\n                        CommonErrorCodeDeprecated.READER_OPERATION_FAILED,\n                        \"Failed to start the split reader thread.\",\n                        e);\n            }\n        }\n    }\n\n    protected PulsarSplitReaderThread createPulsarSplitReaderThread(PulsarPartitionSplit split) {\n        PulsarConsumerMetadata metadata = resolveConsumerMetadata(resolveTablePath(split));\n        return new PulsarSplitReaderThread(\n                this,\n                split,\n                pulsarClient,\n                metadata.getConsumerConfig(),\n                pollTimeout,\n                pollInterval,\n                metadata.getStartCursor(),\n                handover);","sourceCodeStart":181,"sourceCodeEnd":217,"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#L181-L217","documentation":"PulsarSourceReader.addSplits creates and starts a PulsarSplitReaderThread for each new split. Opening the underlying Pulsar consumer can throw PulsarClientException, which is wrapped as READER_OPERATION_FAILED with this message. Data reading cannot start for the affected split because the Pulsar consumer could not be created/started.","triggerScenarios":"Assigning a split whose consumer open() fails with PulsarClientException — broker unreachable at service.url, subscription does not exist in Exclusive/Failover mode conflict, authentication failure, or topic deleted.","commonSituations":"Wrong service.url or port (6650 vs admin 8080); subscription already connected from another client (exclusive consumer conflict); missing subscription when start.mode = SUBSCRIPTION; TLS/auth misconfiguration.","solutions":["Verify service.url (pulsar://host:6650) is reachable and the broker is up; check the wrapped PulsarClientException cause.","Ensure the subscription.name exists or that start.mode/SUBSCRIPTION setup creates it; resolve exclusive-subscription conflicts with other consumers.","Check Pulsar authentication/authorization (token, TLS) configuration on both client and broker side.","Confirm the topic (and partitions) still exist; recreate deleted topics or update the config."],"exampleFix":"// before\nservice.url = \"pulsar://localhost:8080\"\n// after\nservice.url = \"pulsar://localhost:6650\"","handlingStrategy":"try-catch","validationCode":"// pre-flight connectivity check\ntry (PulsarClient c = PulsarClient.builder().serviceUrl(serviceUrl).build()) { /* ok */ }\n// plus: ensure subscription exists\nadmin.topics().getSubscriptions(topic).contains(subscriptionName);","typeGuard":null,"tryCatchPattern":"try {\n    reader.addSplits(splits);\n} catch (PulsarConnectorException e) {\n    if (e.getErrorCode() == CommonErrorCodeDeprecated.READER_OPERATION_FAILED\n            && e.getCause() instanceof PulsarClientException) {\n        // retry with backoff; check broker availability and exclusive-subscription conflicts\n    }\n    throw e;\n}","preventionTips":["Use pulsar://host:6650 for service.url, not the admin HTTP port.","Pre-create the subscription or ensure start.mode setup creates it.","Avoid two exclusive consumers on the same subscription.","Verify TLS/auth settings match broker configuration."],"tags":["pulsar","consumer","split-reader","connection"],"backgroundTag":"connection-refused","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}