tencent cloud

Consumer<byte[]> consumer = client.newConsumer()
    .topic("persistent://1******30/my-ns/topic1")
    .subscriptionType(SubscriptionType.Shared)// Only the shared consumption mode supports the retry and dead letter mechanisms
    .enableRetry(true)
    .subscriptionName("sub1")
    .subscribe();

Consumer<byte[]> consumer = pulsarClient.newConsumer()
    .topic("persistent://pulsar-****")
    .subscriptionName("sub1")
    .subscriptionType(SubscriptionType.Shared)
    .enableRetry(true)// Enable consumption retry
    .deadLetterPolicy(DeadLetterPolicy.builder()
          .maxRedeliverCount(maxRedeliveryCount)// Specify the maximum number of retries
          .retryLetterTopic("persistent://my-property/my-ns/sub1-retry")// Specify the retry letter topic
          .deadLetterTopic("persistent://my-property/my-ns/sub1-dlq")// Specify the dead letter topic
          .build())
    .subscribe();

 // Specify any delay time
 consumer.reconsumeLater(msg, 1000L, TimeUnit.MILLISECONDS);
 // Specify the delay level
 consumer.reconsumeLater(msg, 1);
 // Increase with level
 consumer.reconsumeLater(msg);

{
  REAL_TOPIC="persistent://my-property/my-ns/test, 
  ORIGIN_MESSAGE_ID=314:28:-1, 
  RETRY_TOPIC="persistent://my-property/my-ns/my-subscription-retry, 
  RECONSUMETIMES=16
}

Original consumption: msgid=1:1:0:1
1st retry: msgid=2:1:-1
2nd retry: msgid=2:2:-1
3rd retry: msgid=2:3:-1
.......
16th retry: msgid=2:16:0:1
17th write to the dead letter topic: msgid=3:1:-1

Consumer<byte[]> consumer1 = client.newConsumer()
        .topic("persistent://pulsar-****")
        .subscriptionName("my-subscription")
        .subscriptionType(SubscriptionType.Shared)
        .enableRetry(true)// Enable consumption retry
        //.deadLetterPolicy(DeadLetterPolicy.builder()
        //         .maxRedeliverCount(maxRedeliveryCount)
        //         .retryLetterTopic("persistent://my-property/my-ns/my-subscription-retry")// Specify the retry letter topic
        //         .deadLetterTopic("persistent://my-property/my-ns/my-subscription-dlq")// Specify the dead letter topic
        //         .build())
        .subscribe();

while (true) {
      Message msg = consumer.receive();
      try {
            // Do something with the message
            System.out.printf("Message received: %s", new String(msg.getData()));
            // Acknowledge the message so that it can be deleted by the message broker
            consumer.acknowledge(msg);
      } catch (Exception e){
            // select reconsume policy
            consumer.reconsumeLater(msg, 1000L, TimeUnit.MILLISECONDS);
            //consumer.reconsumeLater(msg, 1);
            //consumer.reconsumeLater(msg);
      }
}

while (true) {
    Message msg = consumer.receive();
    try {
        // Do something with the message
        System.out.printf("Message received: %s", new String(msg.getData()));
        // Acknowledge the message so that it can be deleted by the message broker
        consumer.acknowledge(msg);
    } catch (Exception e){
        // Message failed to process, redeliver later
        consumer.negativeAcknowledge(msg);
    }
}