[GitHub] [pulsar] bigbang489 edited a comment on issue #6960: Duplicated messages are sent to dead letter topic

[GitBox <gi...@apache.org>]

bigbang489 edited a comment on issue #6960:
URL: https://github.com/apache/pulsar/issues/6960#issuecomment-632557593


   This is my code
   
   `package com.test;
   
   import java.util.concurrent.Executor;
   import java.util.concurrent.Executors;
   import java.util.concurrent.TimeUnit;
   
   import org.apache.pulsar.client.api.AuthenticationFactory;
   import org.apache.pulsar.client.api.ClientBuilder;
   import org.apache.pulsar.client.api.Consumer;
   import org.apache.pulsar.client.api.DeadLetterPolicy;
   import org.apache.pulsar.client.api.Message;
   import org.apache.pulsar.client.api.MessageListener;
   import org.apache.pulsar.client.api.Producer;
   import org.apache.pulsar.client.api.PulsarClient;
   import org.apache.pulsar.client.api.PulsarClientException;
   import org.apache.pulsar.client.api.SubscriptionType;
   
   public class App {
   	static Executor executor = Executors.newFixedThreadPool(3);
   	
   	public static void main(String[] args) throws InterruptedException, PulsarClientException {
   		startConsumer("my-topic", false);
   		startConsumer("my-topic", false);
   		startConsumer("my-topic", false);
   		startConsumer("DLQ", true); // This consumer is for dead letter
   		Thread.sleep(2000);
   		publish(String.valueOf(System.currentTimeMillis()), "my-topic");
   	}
   	static void startConsumer(String topic, boolean ack) {
   		executor.execute(()-> {
   			try {
   				subscribe(topic, ack);
   			} catch (PulsarClientException e) {
   				e.printStackTrace();
   			}
   		});
   	}
   	
   	static Consumer<byte[]> subscribe(String topic, boolean ack) throws PulsarClientException {
   		ClientBuilder cb = PulsarClient.builder();
   		cb.serviceUrl("pulsar://pulsar:6650");
   		cb.authentication(AuthenticationFactory.token("*************"));
   		PulsarClient client = cb.build();
   		@SuppressWarnings("serial")
   		Consumer<byte[]> consumer = client.newConsumer()
   			.subscriptionName("test")
   			.subscriptionType(SubscriptionType.Shared)
   			.topic(topic)
   			.ackTimeout(12, TimeUnit.SECONDS)
   			.messageListener(new MessageListener<byte[]>() {
   
   				@Override
   				public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
   					System.out.println("\r\nReceive messaged: '" + new String(msg.getData()) + "' from topic: " + msg.getTopicName());
   					try {
   						if(ack) {
   							consumer.acknowledge(msg);
   						}
   					} catch (PulsarClientException e) {
   						e.printStackTrace();
   					}
   				}
   				
   			})
   			.deadLetterPolicy(DeadLetterPolicy.builder()
   					.deadLetterTopic("DLQ")
   					.maxRedeliverCount(2)
   					.build())
   			.subscribe();
   		System.out.println("\r\nStarted a consumer...");
   		return consumer;
   	}
   	
   	static void publish(String message, String topic) throws PulsarClientException {
   		ClientBuilder cb = PulsarClient.builder();
   		cb.serviceUrl("pulsar://pulsar:6650");
   		cb.authentication(AuthenticationFactory.token("***************"));
   		PulsarClient client = cb.build();
   		Producer<byte[]> producer = client.newProducer().topic(topic).create();
   		try {
   			System.out.println("\r\nSent message: " + producer.send(message.getBytes()));
   		}
   		finally {
   			producer.close();
   		}
   		
   	}
   }
   `
   
   POM file:
   `<dependency>
     		<groupId>org.apache.pulsar</groupId>
     		<artifactId>pulsar-client</artifactId>
     		<version>2.5.1</version>
     	</dependency>
     	<dependency>
     		<groupId>org.apache.pulsar</groupId>
     		<artifactId>pulsar-client-api</artifactId>
     		<version>2.5.1</version>
     	</dependency>`
   
   **The result** 
   Receive messaged: '1590133963111' from topic: persistent://public/default/my-topic
   
   Receive messaged: '1590133963111' from topic: persistent://public/default/my-topic
   
   Receive messaged: '1590133963111' from topic: persistent://public/default/my-topic
   
   **Receive messaged: '1590133963111' from topic: persistent://public/default/DLQ**
   
   Receive messaged: '1590133963111' from topic: persistent://public/default/my-topic
   
   **Receive messaged: '1590133963111' from topic: persistent://public/default/DLQ**
   
   As you see, the message is sent to DLQ twice.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org