{"record":{"id":"9abcac740f1cef59","repo":"apache/pulsar","slug":"recordsequence-needs-to-be-specified-for-every-rec","errorCode":null,"errorMessage":"RecordSequence needs to be specified for every record while in Effectively-once mode","messagePattern":"RecordSequence needs to be specified for every record while in Effectively-once mode","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java","lineNumber":225,"sourceCode":"                // we must use the destination topic schema\n                schemaToWrite = schema;\n            }\n            String topicName = record.getDestinationTopic().orElse(pulsarSinkConfig.getTopic());\n            String partitionId = record.getPartitionId().get();\n            String producerName = partitionId;\n            Producer<T> producer = getProducer(topicName, schemaToWrite, producerName, partitionId);\n            if (schemaToWrite != null) {\n                return producer.newMessage(schemaToWrite);\n            } else {\n                return producer.newMessage();\n            }\n        }\n\n        @Override\n        public void sendOutputMessage(TypedMessageBuilder<T> msg, AbstractSinkRecord<T> record) {\n\n            if (!record.getRecordSequence().isPresent()) {\n                throw new RuntimeException(\n                        \"RecordSequence needs to be specified for every record while in Effectively-once mode\");\n            }\n\n            // assign sequence id to output message for idempotent producing\n            msg.sequenceId(record.getRecordSequence().get());\n            CompletableFuture<MessageId> future = msg.sendAsync();\n\n            future.thenAccept(messageId -> record.ack()).exceptionally(getPublishErrorHandler(record, true));\n        }\n    }\n\n    public PulsarSink(PulsarClient client, PulsarSinkConfig pulsarSinkConfig, Map<String, String> properties,\n                      ComponentStatsManager stats, ClassLoader functionClassLoader, ProducerCache producerCache) {\n        this.client = client;\n        this.pulsarSinkConfig = pulsarSinkConfig;\n        this.topicSchema = new TopicSchema(client, functionClassLoader);\n        this.properties = properties;\n        this.stats = stats;","sourceCodeStart":207,"sourceCodeEnd":243,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java#L207-L243","documentation":"Under EFFECTIVELY_ONCE, the sink assigns the record's sequence to the produced message so retries are idempotent. Thrown by sendOutputMessage when record.getRecordSequence() is empty, since deduplication of the output message is impossible without a sequence number.","triggerScenarios":"Producing to a sink with EFFECTIVELY_ONCE while the incoming Record lacks a record sequence (Optional.empty from getRecordSequence()).","commonSituations":"Custom source records that don't supply Record.getRecordSequence(); switching a topic without sequence metadata (e.g. non-Pulsar-origin records) into effectively-once mode.","solutions":["Have the source's Record implement getRecordSequence() returning a monotonically meaningful sequence.","When the record originates from a Pulsar topic, propagate the source message's sequence id into the record.","Otherwise switch the sink to ATLEAST_ONCE processing guarantees."],"exampleFix":"// before\npublic Optional<Long> getRecordSequence() { return Optional.empty(); }\n// after\npublic Optional<Long> getRecordSequence() { return Optional.of(sourceMessage.getSequenceId()); }","handlingStrategy":"type-guard","validationCode":"if (config.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE\n    && !record.getRecordSequence().isPresent()) {\n  throw new IllegalStateException(\"recordSequence required for EFFECTIVELY_ONCE\");\n}","typeGuard":"boolean hasRecordSequence(Record<?> r) {\n  return r.getRecordSequence() != null && r.getRecordSequence().isPresent();\n}","tryCatchPattern":"try {\n  sink.write(record);\n} catch (RuntimeException e) {\n  if (e.getMessage().contains(\"RecordSequence needs to be specified\")) {\n    // supply sequence ids in the source Record or downgrade guarantee\n  }\n}","preventionTips":["Propagate the source Pulsar message sequenceId into custom Records","Use EFFECTIVELY_ONCE only with sequence-capable sources","Add pipeline tests that exercise the fail path with sequenceless records"],"tags":["effectively-once","sink","idempotency","pulsar-functions"],"backgroundTag":"missing-record-sequence","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}