apache/pulsar · error · RuntimeException

Failed to process message: ${messageId}

Error message

Failed to process message: ${messageId}

What it means

PulsarSource.buildRecord attaches a failFunction to the record; when the record fails to process under EFFECTIVELY_ONCE guarantees, negative acknowledgement is not used (it breaks exactly-once), so it throws instead. The exception surfaces the message id that failed processing.

Source

Thrown at pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java:157

                        consumer.acknowledgeCumulativeAsync(message)
                                .whenComplete((unused, throwable) -> message.release());
                    } else {
                        consumer.acknowledgeAsync(message).whenComplete((unused, throwable) -> message.release());
                    }
                })
                .ackFunction(() -> {
                    if (pulsarSourceConfig
                            .getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
                        consumer.acknowledgeCumulativeAsync(message)
                                .whenComplete((unused, throwable) -> message.release());
                    } else {
                        consumer.acknowledgeAsync(message).whenComplete((unused, throwable) -> message.release());
                    }
                }).failFunction(() -> {
                    try {
                        if (pulsarSourceConfig.getProcessingGuarantees()
                                == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
                            throw new RuntimeException("Failed to process message: " + message.getMessageId());
                        }
                        consumer.negativeAcknowledge(message);
                    } finally {
                        // don't need to check if message pooling is set
                        // client will automatically check
                        message.release();
                    }
                })
                .build();
    }

    @SuppressWarnings("unchecked") // schema type is determined by runtime type arg
    protected PulsarSourceConsumerConfig<T> buildPulsarSourceConsumerConfig(String topic, ConsumerConfig conf,
                                                                            Class<?> typeArg) {
        PulsarSourceConsumerConfig.PulsarSourceConsumerConfigBuilder<T> consumerConfBuilder =
                PulsarSourceConsumerConfig.<T>builder().isRegexPattern(conf.isRegexPattern())
                        .receiverQueueSize(conf.getReceiverQueueSize())
                        .consumerProperties(conf.getConsumerProperties());

View on GitHub (pinned to 820761864e)

Solutions

1. Fix the user function so the offending message processes correctly (the message id in the message identifies the poison input).
2. Handle/validate malformed records inside the function instead of letting them fail processing.
3. Switch to ATLEAST_ONCE if negative-ack redelivery semantics are desired.

Example fix

// before: function throws on bad input -> failFunction -> RuntimeException
// after: catch and handle in user function
try { process(input); } catch (IllegalArgumentException e) { log.warn("skipping bad record", e); }

// inside the user function's process()
try {
  output = process(inputRecord);
} catch (Exception e) {
  log.error("failed processing messageId {}", inputRecord.getMessageId(), e);
  throw e; // under EFFECTIVELY_ONCE this surfaces as 'Failed to process message'
}

When it happens

Trigger: A function/sink processing failure triggers the record's failFunction while pulsarSourceConfig.getProcessingGuarantees() == EFFECTIVELY_ONCE, causing the RuntimeException instead of consumer.negativeAcknowledge(message).

Common situations: Function code repeatedly failing on a specific message; running effectively-once pipelines where a single poison pill message crashes the instance instead of being redelivered via negative ack.

Related errors

• The record returned by the source cannot be null
• Source does not implement correct interface
• PartitionId needs to be specified for every record while in
• RecordSequence needs to be specified for every record while
• Partitioned topic is not available in effectively_once mode.

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