{"record":{"id":"54c4ed7a1ec543bb","repo":"apache/pulsar","slug":"partitionid-needs-to-be-specified-for-every-record","errorCode":null,"errorMessage":"PartitionId needs to be specified for every record while in Effectively-once mode","messagePattern":"PartitionId 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":200,"sourceCode":"                    .thenAccept(messageId -> record.ack())\n                    .exceptionally(getPublishErrorHandler(record, true));\n        }\n    }\n\n    @VisibleForTesting\n    class PulsarSinkManualProcessor extends PulsarSinkAtMostOnceProcessor {\n        @Override\n        public void sendOutputMessage(TypedMessageBuilder<T> msg, AbstractSinkRecord<T> record) {\n            super.sendOutputMessage(msg, record);\n        }\n    }\n\n    @VisibleForTesting\n    class PulsarSinkEffectivelyOnceProcessor extends PulsarSinkProcessorBase {\n        @Override\n        public TypedMessageBuilder<T> newMessage(AbstractSinkRecord<T> record) {\n            if (!record.getPartitionId().isPresent()) {\n                throw new RuntimeException(\n                        \"PartitionId needs to be specified for every record while in Effectively-once mode\");\n            }\n            Schema<T> schemaToWrite = record.getSchema();\n            if (!record.shouldSetSchema()) {\n                // we are receiving data directly from another Pulsar topic\n                // and the Function return type is not a Record\n                // 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            }","sourceCodeStart":182,"sourceCodeEnd":218,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java#L182-L218","documentation":"In EFFECTIVELY_ONCE processing mode, PulsarSink's PulsarSinkEffectivelyOnceProcessor builds idempotent output messages, which requires each record to carry a partition id. Thrown by newMessage when record.getPartitionId() is absent, because the sink cannot construct a deterministic message key/producer route for exactly-once semantics.","triggerScenarios":"A custom Source Record implementation that does not set partitionId(), feeding a sink configured with ProcessingGuarantees=EFFECTIVELY_ONCE.","commonSituations":"Custom user sources (not PulsarSource) that return Records without partition information; moving a pipeline from ATLEAST_ONCE to EFFECTIVELY_ONCE without updating the source's Record implementation.","solutions":["Make the source's Record return a non-empty partitionId via getPartitionId().","If partition semantics don't apply, downgrade the sink's processing guarantee to ATLEAST_ONCE.","Use the built-in PulsarSource, which populates partition id automatically for partitioned topics."],"exampleFix":"// before\npublic class MyRecord<T> implements Record<T> {\n  public Optional<String> getPartitionId() { return Optional.empty(); }\n}\n// after\npublic Optional<String> getPartitionId() { return Optional.of(\"0\"); }","handlingStrategy":"type-guard","validationCode":"if (sinkConfig.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE\n    && !record.getPartitionId().isPresent()) {\n  throw new IllegalStateException(\"partitionId required for EFFECTIVELY_ONCE\");\n}","typeGuard":"boolean hasPartitionId(Record<?> r) {\n  return r.getPartitionId() != null && r.getPartitionId().isPresent();\n}","tryCatchPattern":"try {\n  sink.write(record);\n} catch (RuntimeException e) {\n  if (e.getMessage().contains(\"PartitionId needs to be specified\")) {\n    // fix the source Record implementation or change processing guarantee\n  }\n}","preventionTips":["Implement getPartitionId() in all custom Record classes","Only enable EFFECTIVELY_ONCE with sources that supply partition ids","Test custom sources under effectively-once mode before production"],"tags":["effectively-once","sink","record","pulsar-functions"],"backgroundTag":"missing-partition-id","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}