{"record":{"id":"374843510a7c4622","repo":"apache/seatunnel","slug":"pulsarconnectorerrorcode-get-last-cursor-failed","errorCode":"PulsarConnectorErrorCode.GET_LAST_CURSOR_FAILED","errorMessage":"Failed to get the last cursor","messagePattern":"Failed to get the last cursor","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/enumerator/cursor/stop/LatestMessageStopCursor.java","lineNumber":46,"sourceCode":"import org.apache.pulsar.client.api.MessageId;\n\n/**\n * A stop cursor that initialize the position to the latest message id. The offsets initialization\n * are taken care of by the {@code PulsarPartitionSplitReaderBase} instead of by the {@code\n * PulsarSourceEnumerator}.\n */\npublic class LatestMessageStopCursor implements StopCursor {\n    private static final long serialVersionUID = 1L;\n\n    private MessageId messageId;\n\n    public void prepare(PulsarAdmin admin, TopicPartition partition) {\n        if (messageId == null) {\n            String topic = partition.getFullTopicName();\n            try {\n                messageId = admin.topics().getLastMessageId(topic);\n            } catch (PulsarAdminException e) {\n                throw new PulsarConnectorException(\n                        PulsarConnectorErrorCode.GET_LAST_CURSOR_FAILED,\n                        \"Failed to get the last cursor\",\n                        e);\n            }\n        }\n    }\n\n    @Override\n    public boolean shouldStop(Message<?> message) {\n        MessageId id = message.getMessageId();\n        return id.compareTo(messageId) >= 0;\n    }\n\n    @Override\n    public StopCursor copy() {\n        return new LatestMessageStopCursor();\n    }\n}","sourceCodeStart":28,"sourceCodeEnd":64,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/enumerator/cursor/stop/LatestMessageStopCursor.java#L28-L64","documentation":"LatestMessageStopCursor.prepare fetches the topic's last message ID via the PulsarAdmin client so the split knows where to stop. If admin.topics().getLastMessageId(topic) throws a PulsarAdminException, it is wrapped as GET_LAST_CURSOR_FAILED with this message. Reading fails at split-prepare time because the stop position cannot be determined.","triggerScenarios":"Using stop.mode = LATEST when PulsarAdmin.topics().getLastMessageId fails for the full topic name — broker unreachable, topic does not exist, permission denied, or topic name (partitioned topic suffix) is malformed.","commonSituations":"Misconfigured admin.service-url or authentication; typo in topic name; reading a partitioned topic where the partition suffix handling mismatches; topic deleted between discovery and reader start.","solutions":["Verify admin.service-url and Pulsar admin credentials are correct and reachable from the worker.","Confirm the topic exists and the configured topic name matches exactly (including persistent:// domain/tenant/namespace/topic).","Check broker logs for the underlying PulsarAdminException cause (TopicNotFound, NotAuthorized, etc.).","For partitioned topics, ensure the partition reference used matches an existing partition."],"exampleFix":"// before\nadmin.service-url = \"http://broker:8081\"\n// after (correct admin port/service URL)\nadmin.service-url = \"http://broker:8080\"","handlingStrategy":"try-catch","validationCode":"// pre-flight: topic must exist and admin reachable\nadmin.namespaces().getTopics(namespace).contains(fullTopicName);","typeGuard":null,"tryCatchPattern":"try {\n    prepareStopCursor(admin, partition);\n} catch (PulsarConnectorException e) {\n    if (e.getErrorCode() == PulsarConnectorErrorCode.GET_LAST_CURSOR_FAILED) {\n        // check cause: TopicNotFound -> fix topic name; auth -> fix credentials\n    }\n    throw e;\n}","preventionTips":["Validate topic names exactly (persistent://tenant/namespace/topic).","Confirm admin.service-url and credentials before running LATEST stop mode jobs.","Pre-create topics so getLastMessageId never hits TopicNotFound."],"tags":["pulsar","admin-client","cursor","stop-position"],"backgroundTag":"http-request-failed","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"}