apache/pulsar · error · RuntimeException

RecordSequence needs to be specified for every record while

Error message

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

What it means

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.

Source

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

                // 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();
            }
        }

        @Override
        public void sendOutputMessage(TypedMessageBuilder<T> msg, AbstractSinkRecord<T> record) {

            if (!record.getRecordSequence().isPresent()) {
                throw new RuntimeException(
                        "RecordSequence needs to be specified for every record while in Effectively-once mode");
            }

            // assign sequence id to output message for idempotent producing
            msg.sequenceId(record.getRecordSequence().get());
            CompletableFuture<MessageId> future = msg.sendAsync();

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

    public PulsarSink(PulsarClient client, PulsarSinkConfig pulsarSinkConfig, Map<String, String> properties,
                      ComponentStatsManager stats, ClassLoader functionClassLoader, ProducerCache producerCache) {
        this.client = client;
        this.pulsarSinkConfig = pulsarSinkConfig;
        this.topicSchema = new TopicSchema(client, functionClassLoader);
        this.properties = properties;
        this.stats = stats;

View on GitHub (pinned to 820761864e)

Solutions

1. Have the source's Record implement getRecordSequence() returning a monotonically meaningful sequence.
2. When the record originates from a Pulsar topic, propagate the source message's sequence id into the record.
3. Otherwise switch the sink to ATLEAST_ONCE processing guarantees.

Example fix

// before
public Optional<Long> getRecordSequence() { return Optional.empty(); }
// after
public Optional<Long> getRecordSequence() { return Optional.of(sourceMessage.getSequenceId()); }

if (config.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE
    && !record.getRecordSequence().isPresent()) {
  throw new IllegalStateException("recordSequence required for EFFECTIVELY_ONCE");
}

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

try {
  sink.write(record);
} catch (RuntimeException e) {
  if (e.getMessage().contains("RecordSequence needs to be specified")) {
    // supply sequence ids in the source Record or downgrade guarantee
  }
}

When it happens

Trigger: Producing to a sink with EFFECTIVELY_ONCE while the incoming Record lacks a record sequence (Optional.empty from getRecordSequence()).

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

Related errors

• PartitionId needs to be specified for every record while in
• SourceRecord class type must be PulsarRecord
• Sink does not implement correct interface
• Failed to process message: ${messageId}
• Partitioned topic is not available in effectively_once mode.

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