negativeAcknowledge() parameter does not take effect

[yebai1105]

Expected behavior

message redelivery

Actual behavior

Message cannot be redelivered

Steps to reproduce

producer code：

public static void main(String[] args) throws PulsarClientException, InterruptedException {
        String token = "xxx";
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("xxx")
                .authentication(AuthenticationFactory.token(token))
                .memoryLimit(64, SizeUnit.MEGA_BYTES)
                .build();
        Producer<String> producer = client.newProducer(Schema.STRING)
                .topic("xxx")
                .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
                .enableBatching(true)
                .batchingMaxBytes(1048576)
                .batchingMaxMessages(10000)
                .batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS)
                .blockIfQueueFull(true)
                .create();

        long i = 0;
        while (true) {
            i++;
            producer.sendAsync("Message " + i).thenAccept(msgId -> {
                System.out.println("Message " + msgId + " successfully sent");
            });
            if(i>10)break;
        }
    }

consumer code：

public static void main(String[] args) throws PulsarClientException {

        String token = "xxx";
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("xxx")
                .authentication(AuthenticationFactory.token(token))
                .memoryLimit(64, SizeUnit.MEGA_BYTES)
                .build();


        Consumer<String> consumer = client.newConsumer(Schema.STRING)
                .topic("xxx")
                .subscriptionName("xxx")
                .subscriptionType(SubscriptionType.Exclusive)
                .receiverQueueSize(1000)
                .batchReceivePolicy(BatchReceivePolicy.builder()
                        .maxNumMessages(1000)
                        .maxNumBytes(1024 * 1024 * 10)
                        .timeout(10, TimeUnit.MILLISECONDS)
                        .build())
                .subscribe();

        while (true) {
            Messages<String> msgs = consumer.batchReceive();
            msgs.forEach(obj -> {
                Message msg = null;
                try {
                    msg = (Message) obj;
                    System.out.println("Message received: " + msg.getValue());
                    consumer.acknowledge(msg);
                } catch (Exception e) {
                    // Message processing failed, resend later

                }finally {
                    //Test: After each confirmation, cancel the confirmation again to make the message redelivery
                    consumer.negativeAcknowledge(msg);
                }
            });
        }
    }

use "consumer.negativeAcknowledge(msg)" or "consumer.negativeAcknowledge(new MessageIdImpl(x,x,x))" After executing the consumer program, no message redelivery was found

System configuration

Pulsar version: 2.9.2/2.8.2

[codelipenghui]

@yebai1105 You can acked the message at

msg = (Message) obj;
System.out.println("Message received: " + msg.getValue());
consumer.acknowledge(msg);

A acked message can not be redeliver again

[yebai1105]

Unacknowledged messages will be redelivered by default. If A acked message can not be redeliver again, what is the role of "negativeAcknowledge"?In what scenarios is the "negativeAcknowledge" parameter used?

[drriguz]

I'm also confused about this, I found that messages are not redelivered in my application by using the following code:

try {
            MetaDataChangedEvent<?> message = messageDecoder.decodeEvent(msg.getData());
            if (message instanceof AppDescriptionChanged) {
                messageProcessService.processEvent((AppDescriptionChanged) message);
            } else {
                messageProcessService.processEvent((SnapshotEvent<?>) message);
            }
            consumer.acknowledge(msg);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("Failed to process message {}: interrupted", msg.getMessageId(), e);
            consumer.negativeAcknowledge(msg);
        } catch (Exception e) {
            // todo: error handling & impl retry strategy
            logger.error("Failed to process message {}", msg.getMessageId(), e);
            consumer.negativeAcknowledge(msg);
        }

The document does not say clearly about that, and there's also someone asking in Stackoverflow: https://stackoverflow.com/questions/69590720/pulsar-if-a-message-gets-nackd-negativeacknowledge-when-will-it-be-redeliv

And also, the document mentioned that automatic retry is disabled by default, is this the reason that negative-acked messages are not redelivered?

By default, automatic retry is disabled. You can set enableRetry to true to enable automatic retry on the consumer.

[github-actions]

The issue had no activity for 30 days, mark with Stale label.