What causes "Failed to process message: ${messageId}" in apache/pulsar?

Fix the user function so the offending message processes correctly (the message id in the message identifies the poison input).

apache/pulsar · error · RuntimeException

Failed to process message

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.

Solutions

Fix the user function so the offending message processes correctly (the message id in the message identifies the poison input).
Handle/validate malformed records inside the function instead of letting them fail processing.
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); }

Defensive patterns

Strategy: try-catch

Try / catch

// 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'
}

Prevention

Validate/repair input records inside the function rather than failing
Log the failing message id to inspect the poison message on the topic
Prefer ATLEAST_ONCE if negative-ack redelivery is desired for bad messages

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

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

Appendix: 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)