apache/pulsar · error · RuntimeException

PartitionId needs to be specified for every record while in

Error message

PartitionId needs to be specified for every record while in Effectively-once mode

What it means

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.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java:200

                    .thenAccept(messageId -> record.ack())
                    .exceptionally(getPublishErrorHandler(record, true));
        }
    }

    @VisibleForTesting
    class PulsarSinkManualProcessor extends PulsarSinkAtMostOnceProcessor {
        @Override
        public void sendOutputMessage(TypedMessageBuilder<T> msg, AbstractSinkRecord<T> record) {
            super.sendOutputMessage(msg, record);
        }
    }

    @VisibleForTesting
    class PulsarSinkEffectivelyOnceProcessor extends PulsarSinkProcessorBase {
        @Override
        public TypedMessageBuilder<T> newMessage(AbstractSinkRecord<T> record) {
            if (!record.getPartitionId().isPresent()) {
                throw new RuntimeException(
                        "PartitionId needs to be specified for every record while in Effectively-once mode");
            }
            Schema<T> schemaToWrite = record.getSchema();
            if (!record.shouldSetSchema()) {
                // we are receiving data directly from another Pulsar topic
                // and the Function return type is not a Record
                // we must use the destination topic schema
                schemaToWrite = schema;
            }
            String topicName = record.getDestinationTopic().orElse(pulsarSinkConfig.getTopic());
            String partitionId = record.getPartitionId().get();
            String producerName = partitionId;
            Producer<T> producer = getProducer(topicName, schemaToWrite, producerName, partitionId);
            if (schemaToWrite != null) {
                return producer.newMessage(schemaToWrite);
            } else {
                return producer.newMessage();
            }

View on GitHub (pinned to 820761864e)

Solutions

1. Make the source's Record return a non-empty partitionId via getPartitionId().
2. If partition semantics don't apply, downgrade the sink's processing guarantee to ATLEAST_ONCE.
3. Use the built-in PulsarSource, which populates partition id automatically for partitioned topics.

Example fix

// before
public class MyRecord<T> implements Record<T> {
  public Optional<String> getPartitionId() { return Optional.empty(); }
}
// after
public Optional<String> getPartitionId() { return Optional.of("0"); }

if (sinkConfig.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE
    && !record.getPartitionId().isPresent()) {
  throw new IllegalStateException("partitionId required for EFFECTIVELY_ONCE");
}

boolean hasPartitionId(Record<?> r) {
  return r.getPartitionId() != null && r.getPartitionId().isPresent();
}

try {
  sink.write(record);
} catch (RuntimeException e) {
  if (e.getMessage().contains("PartitionId needs to be specified")) {
    // fix the source Record implementation or change processing guarantee
  }
}

When it happens

Trigger: A custom Source Record implementation that does not set partitionId(), feeding a sink configured with ProcessingGuarantees=EFFECTIVELY_ONCE.

Common situations: 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.

Related errors

• RecordSequence needs to be specified for every record while
• SourceRecord class type must be PulsarRecord
• The value in the record returned by the source cannot be nul
• Sink does not implement correct interface
• Failed to process message: ${messageId}

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/54c4ed7a1ec543bb. Report an issue: GitHub.