{"record":{"id":"13d3e5012083cd9a","repo":"apache/pulsar","slug":"non-null-message-is-required","errorCode":null,"errorMessage":"Non-null message is required","messagePattern":"Non-null message is required","errorType":"exception","errorClass":"PulsarClientException.InvalidMessageException","httpStatus":null,"severity":"error","filePath":"pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java","lineNumber":425,"sourceCode":"                break;\n            }\n            if (!opBatchReceive.future.isDone()) {\n                opBatchReceive.future.completeExceptionally(\n                        new PulsarClientException.AlreadyClosedException(\n                                String.format(\"The consumer which subscribes the topic %s with subscription name %s was\"\n                                                + \" already closed when cleaning and closing the consumers\",\n                                        topic, subscription)));\n            }\n        }\n    }\n\n    protected abstract Messages<T> internalBatchReceive() throws PulsarClientException;\n\n    protected abstract CompletableFuture<Messages<T>> internalBatchReceiveAsync();\n\n    private static void validateMessageId(Message<?> message) throws PulsarClientException {\n        if (message == null) {\n            throw new PulsarClientException.InvalidMessageException(\"Non-null message is required\");\n        }\n        if (message.getMessageId() == null) {\n            throw new PulsarClientException.InvalidMessageException(\"Cannot handle message with null messageId\");\n        }\n    }\n\n    private static void validateMessageId(MessageId messageId) throws PulsarClientException {\n        if (messageId == null) {\n            throw new PulsarClientException.InvalidMessageException(\"Cannot handle message with null messageId\");\n        }\n    }\n\n    private static void validateMessageIds(List<MessageId> messageIdList) throws PulsarClientException {\n        if (messageIdList == null) {\n            throw new PulsarClientException.InvalidMessageException(\"Cannot handle messages with null messageIdList\");\n        }\n        for (MessageId messageId : messageIdList) {\n            validateMessageId(messageId);","sourceCodeStart":407,"sourceCodeEnd":443,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java#L407-L443","documentation":"ConsumerBase.validateMessageId(Message) rejects a null Message argument before any acknowledgment-style operation is attempted, throwing InvalidMessageException with 'Non-null message is required'. Ack operations need a real message instance to extract its MessageId, so a null reference is a caller-side programming bug rather than a broker problem.","triggerScenarios":"Calling consumer.acknowledge(null), acknowledgeCumulative(null), acknowledgeAsync(null), reconsumeLaterAsync(null, ...), validateMessageIds/validateMessages with a collection containing null — often when message comes from a map lookup or an Optional that was unwrapped unsafely.","commonSituations":"Caching messages in a Map<MessageId, Message> and acknowledging a missing key; passing the result of a receive() that was short-circuited; processing batched messages where a null element slipped into the list.","solutions":["Null-check the message before calling acknowledge/reconsumeLater","Fix the upstream code that produced the null reference (map miss, Optional.get, uninitialized field)","Use validateMessages/validateMessageIds yourself in tests to fail fast"],"exampleFix":"// before\nconsumer.acknowledge(messageCache.get(id)); // may be null\n// after\nMessage<String> msg = messageCache.get(id);\nif (msg != null) {\n    consumer.acknowledge(msg);\n}","handlingStrategy":"validation","validationCode":"if (message == null) {\n    throw new IllegalArgumentException(\"Cannot ack a null message\");\n}\nconsumer.acknowledge(message);","typeGuard":"boolean isAckable(Message<?> m) {\n    return m != null && m.getMessageId() != null;\n}","tryCatchPattern":"try {\n    consumer.acknowledge(message);\n} catch (PulsarClientException.InvalidMessageException e) {\n    log.warn(\"Invalid message for ack: {}\", e.getMessage());\n}","preventionTips":["Only ack Message objects received directly from this consumer","Null-check results of cache/map lookups before acking","Keep Message references immutable from receive() to ack()"],"tags":["pulsar","null-check","acknowledgment","invalid-message"],"backgroundTag":"null-message-argument","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"}