{"record":{"id":"6c0756d1a6ce198f","repo":"apache/pulsar","slug":"failed-to-process-message-messageid","errorCode":null,"errorMessage":"Failed to process message: ${messageId}","messagePattern":"Failed to process message: (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java","lineNumber":157,"sourceCode":"                        consumer.acknowledgeCumulativeAsync(message)\n                                .whenComplete((unused, throwable) -> message.release());\n                    } else {\n                        consumer.acknowledgeAsync(message).whenComplete((unused, throwable) -> message.release());\n                    }\n                })\n                .ackFunction(() -> {\n                    if (pulsarSourceConfig\n                            .getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {\n                        consumer.acknowledgeCumulativeAsync(message)\n                                .whenComplete((unused, throwable) -> message.release());\n                    } else {\n                        consumer.acknowledgeAsync(message).whenComplete((unused, throwable) -> message.release());\n                    }\n                }).failFunction(() -> {\n                    try {\n                        if (pulsarSourceConfig.getProcessingGuarantees()\n                                == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {\n                            throw new RuntimeException(\"Failed to process message: \" + message.getMessageId());\n                        }\n                        consumer.negativeAcknowledge(message);\n                    } finally {\n                        // don't need to check if message pooling is set\n                        // client will automatically check\n                        message.release();\n                    }\n                })\n                .build();\n    }\n\n    @SuppressWarnings(\"unchecked\") // schema type is determined by runtime type arg\n    protected PulsarSourceConsumerConfig<T> buildPulsarSourceConsumerConfig(String topic, ConsumerConfig conf,\n                                                                            Class<?> typeArg) {\n        PulsarSourceConsumerConfig.PulsarSourceConsumerConfigBuilder<T> consumerConfBuilder =\n                PulsarSourceConsumerConfig.<T>builder().isRegexPattern(conf.isRegexPattern())\n                        .receiverQueueSize(conf.getReceiverQueueSize())\n                        .consumerProperties(conf.getConsumerProperties());","sourceCodeStart":139,"sourceCodeEnd":175,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java#L139-L175","documentation":"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.","triggerScenarios":"A function/sink processing failure triggers the record's failFunction while pulsarSourceConfig.getProcessingGuarantees() == EFFECTIVELY_ONCE, causing the RuntimeException instead of consumer.negativeAcknowledge(message).","commonSituations":"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.","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."],"exampleFix":"// before: function throws on bad input -> failFunction -> RuntimeException\n// after: catch and handle in user function\ntry { process(input); } catch (IllegalArgumentException e) { log.warn(\"skipping bad record\", e); }","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// inside the user function's process()\ntry {\n  output = process(inputRecord);\n} catch (Exception e) {\n  log.error(\"failed processing messageId {}\", inputRecord.getMessageId(), e);\n  throw e; // under EFFECTIVELY_ONCE this surfaces as 'Failed to process message'\n}","preventionTips":["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"],"tags":["effectively-once","source","message-processing","pulsar-functions"],"backgroundTag":"message-processing-failed","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"}