{"record":{"id":"57569ae7e961e68e","repo":"apache/seatunnel","slug":"pulsarconnectorerrorcode-open-pulsar-admin-failed","errorCode":"PulsarConnectorErrorCode.OPEN_PULSAR_ADMIN_FAILED","errorMessage":"Failed to create pulsar consumer:","messagePattern":"Failed to create pulsar consumer:","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/PulsarSplitReaderThread.java","lineNumber":142,"sourceCode":"    public void committingCursor(MessageId offsetsToCommit) throws PulsarClientException {\n        if (consumer == null) {\n            consumer = createPulsarConsumer(split);\n        }\n        consumer.acknowledgeCumulative(offsetsToCommit);\n    }\n\n    /** Create a specified {@link Consumer} by the given split information. */\n    protected Consumer<byte[]> createPulsarConsumer(PulsarPartitionSplit split) {\n        ConsumerBuilder<byte[]> consumerBuilder =\n                PulsarConfigUtil.createConsumerBuilder(pulsarClient, consumerConfig);\n\n        consumerBuilder.topic(split.getPartition().getFullTopicName());\n\n        // Create the consumer configuration by using common utils.\n        try {\n            return consumerBuilder.subscribe();\n        } catch (PulsarClientException e) {\n            throw new PulsarConnectorException(\n                    PulsarConnectorErrorCode.OPEN_PULSAR_ADMIN_FAILED,\n                    \"Failed to create pulsar consumer:\",\n                    e);\n        }\n    }\n\n    /**\n     * Closes the Pulsar consumer while exposing the connector classloader to Pulsar cleanup code.\n     */\n    private void closeConsumer() throws IOException {\n        if (consumer != null) {\n            try {\n                PulsarConfigUtil.runWithConnectorClassLoader(consumer::close);\n            } catch (Exception e) {\n                throw new IOException(\"Failed to close Pulsar consumer.\", e);\n            }\n        }\n    }","sourceCodeStart":124,"sourceCodeEnd":160,"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#L124-L160","documentation":"Thrown by PulsarSplitReaderThread.createPulsarConsumer when consumerBuilder.subscribe() raises a PulsarClientException. The wrapper uses OPEN_PULSAR_ADMIN_FAILED, so despite the name it covers consumer subscription failures: connection, authentication, authorization, or nonexistent topic.","triggerScenarios":"open() or committingCursor() triggers consumer creation and the Pulsar client fails to subscribe: broker unreachable, TLS/auth misconfiguration, topic does not exist (and auto-create disabled), or the subscription name is invalid/already in incompatible use.","commonSituations":"Wrong serviceUrl in Pulsar config; broker down or firewall blocking 6650; JWT/TLS credentials expired or misconfigured; topic deleted while job runs; exclusive subscription already held by another consumer.","solutions":["Read the wrapped PulsarClientException cause for the exact failure (connection, authorization, topic-not-found, etc.).","Verify serviceUrl, authPlugin/authParams, and TLS settings in the connector configuration.","Confirm the topic exists (or enable broker auto-creation) and that the principal has consume permission on it.","Check network connectivity from worker nodes to the Pulsar broker (pulsar://6650) and broker health.","For subscription conflicts, ensure subscription names are unique per job or switch to Shared/Key_Shared subscription type."],"exampleFix":"// before (example fix: topic auto-created expectation)\nsource {\n  Pulsar {\n    serviceUrl = \"pulsar://localhost:6650\"\n    topic = \"nonexistent/topic\"\n  }\n}\n// after\nsource {\n  Pulsar {\n    serviceUrl = \"pulsar://broker:6650\"\n    topic = \"public/default/my-topic\"  // must exist, or enable broker autoCreate\n    auth.params = \"{\"token\":\"<valid-jwt>\"}\"\n  }\n}","handlingStrategy":"validation","validationCode":"// preflight check before submitting the job\nPulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(adminUrl).build();\nif (!admin.topics().getPartitionedTopicList(namespace).contains(topic)) {\n    throw new IllegalArgumentException(\"Topic does not exist: \" + topic);\n}","typeGuard":null,"tryCatchPattern":"try { reader.open(); } catch (PulsarConnectorException e) { LOG.error(\"Subscribe failed: {}\", e.getCause()); if (e.getCause() instanceof PulsarClientException.ConnectionException) { /* retry with backoff */ } }","preventionTips":["Verify serviceUrl, auth and TLS settings before deployment","Ensure the topic exists or broker auto-creation is enabled","Confirm consume permissions for the configured principal","Test network access to pulsar://port from worker nodes"],"tags":["pulsar","consumer","subscription","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-14T11:17:12.474Z"}