intermittent Messages loss while consuming the messages from consumer

[ragaur-tibco]

Hi @lhotari @visortelle

I have a query
when I tried to run the producer and send 10K messages and then stop the producer and after some time I tried to consume the messages with (Dead letter topic feature but topic is not provided only retry parameters were provided) from the consumer some messages was not present and some of them are duplicate can you please provide your inputs?

ConsumerBuilder<String> consumerStringBuilder = client.newConsumer(Schema.STRING)
									        .topic("topic name")
									        .subscriptionName(subscriptionName)
									        .replicateSubscriptionState(false)
									        .subscriptionType(SubscriptionType.Shared);

consumerStringBuilder.negativeAckRedeliveryDelay(60, TimeUnit.SECONDS);
										consumerStringBuilder.deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic(null).build());

Consumer<String> stringConsumer= consumerStringBuilder.subscribe();
Message msg = stringConsumer.receive();

[wallacepeng]

messages should only be retained if there are subscriptions created . from your statement, u first publish messages, then start the consumer, i wonder if lot of messages already deleted ?

[ragaur-tibco]

@lhotari

I have used the below config for consumer
"topic"="topic name"
"subscriptionName"=subscriptionName
"replicateSubscriptionState"=false
"subscriptionType"= SubscriptionType.Shared

"negativeAckRedeliveryDelay" = 60
"maxRedeliverCount"=10
"deadLetterTopic"= null

NOTE: I didn't provide the DLQ but used the below code to receive the messages:

>consumerStringBuilder.negativeAckRedeliveryDelay(60, TimeUnit.SECONDS);

> consumerStringBuilder.deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(10).deadLetterTopic(null).build());

> Consumer<String> stringConsumer= consumerStringBuilder.subscribe();

> Message msg = stringConsumer.receive(); 

@lhotari can you please confirm if above use-case is correct or not?
and can you please confirm do we need to provide the DLQ-topic for above similar use-cases?

NOTE: When the producer and consumer are running simultaneously, the consumer is able to consume all the messages, whereas if the consumer and producer are running separately, the consumer is not able to consume messages there were some messages loss

[lhotari]

@ragaur-tibco please share a runnable example that reproduces a problem. I already provided a good template for you so please do your part.

[lhotari]

You can take a look at https://github.com/apache/pulsar-site/blob/d7f55bfdb2d6f5f19e5774b4a2d36322f13c9a6e/docs/tutorials-redeliver-messages.md to find some best practices.

[ragaur-tibco]

Hi @lhotari

we have two class one is producer.java and another consumer.java

PulsarProducerExample .java

`import org.apache.pulsar.client.api.*;

public class PulsarProducerExample {
    public static void main(String[] args) throws PulsarClientException {
        String serviceUrl = "pulsar://localhost:6650";
        String topicName = "topic-Name"";

        // Create a Pulsar client instance
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .build();

        // Create a producer
        Producer<byte[]> producer = client.newProducer()
                .topic(topicName)
                .create();

        // Send messages
        for (int i = 0; i < 100; i++) {
            String message = "Message " + i;
            producer.send(message.getBytes());
            System.out.println("Sent message: " + message);
        }

        // Close the producer and Pulsar client
        producer.close();
        client.close();
    }
}
`

Consumer.java

`import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;

public class Consumer{
    public static void main(String[] args) throws Exception {
        String serviceUrl = "pulsar://localhost:6650";
        String topicName = "topic-Name";
        String subscriptionName = "subscriptionName";

        PulsarClient client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .build();

        Consumer<byte[]> consumer1 = client.newConsumer()
                .topic(topicName)
                .subscriptionName(subscriptionName)
                .subscriptionType(SubscriptionType.Shared) // Use Shared subscription type
                .subscribe();
        while (true) {
           Message msg = consumer1.receive();
            System.out.println("from consumer 1=============================="+consumer1.receive());
            consumer1.acknowledgeAsync(msg);
        }
    }
}



`

Steps to reproduce:

1. Run consumer and topic will be created by itself
2. check the topic stats one consumer created in subscription name
3. stop the consumer and then run the producer and send 100 messages
4. Run consumer and check the topic stats now you will be able to see that a different consumer with different name and some time with same name created (if in consumerName parameter provided with the same consumer name which is already present) and 100 messages available in previous consumer but not in new consumer. Due to which not able to receive messages

Queries:

1. Is there any api from which we can receive the messages from the specific consumer name
2. Is there any way to fetch the all the messages with is available in subscription from all the consumers

[lhotari]

4. Run consumer and check the topic stats now you will be able to see that a different consumer with different name and some time with same name created (if in consumerName parameter provided with the same consumer name which is already present) and 100 messages available in previous consumer but not in new consumer. Due to which not able to receive messages

how is this step reflected in the above code for reproducing the issue?
I don't see any issue here. The code has a shared subscription and the subscription name is the same. Each consumer must have a unique name if you'd like to have the messages delivered to this "different consumer with different name".

Is there any api from which we can receive the messages from the specific consumer name

No. The consumer name isn't used for receiving messages.

Is there any way to fetch the all the messages with is available in subscription from all the consumers

I don't understand this question. What are you trying accomplish here? For each subscription there's a matching cursor which has runtime state that can be observed with the topic statistics (reference) and internal statistics (reference). Some of the cursor internal state is the so called "mark delete position", "read position" and "individually deleted messages". Deletion means acknowledgement in these concepts. The "mark delete position" is the point where all previous messages have been acknowledged in the cursor (and subscription). For more details you could check this blog post: https://dzone.com/articles/understanding-cursors-tracking-mechanism-in-pulsar .