[GitHub] xiangzilv123 commented on issue #3226: Unclosed consumer after calling closeAsync

[GitBox]

xiangzilv123 commented on issue #3226: Unclosed consumer after calling 
closeAsync
URL: https://github.com/apache/pulsar/issues/3226#issuecomment-449265114
 
 
   
   
   ```java
   package test;
   
   import org.apache.pulsar.client.api.Consumer;
   import org.apache.pulsar.client.api.PulsarClient;
   import org.apache.pulsar.client.api.SubscriptionType;
   
   import java.util.concurrent.TimeUnit;
   
   public class TestMain {
   
       private static void waitTillNextMinute() throws InterruptedException {
           System.out.println("waiting for next minute");
           long m = System.currentTimeMillis() / (60 * 1000);
           do {
               Thread.sleep(1000);
           } while (System.currentTimeMillis() / (60 * 1000) == m);
       }
   
       public static void main(String[] args) throws Exception {
           String url = "pulsar://localhost:6650";
           String topic = "test-topic";
           String subscriptionName = "test-subscription";
   
           PulsarClient pulsarClient = PulsarClient.builder()
                   .serviceUrl(url)
                   .build();
           System.out.println("client created");
   
           while (true) {
               Consumer<byte[]> consumer = pulsarClient.newConsumer()
                       .topic(topic)
                       .ackTimeout(2 * 60, TimeUnit.MINUTES)
                       .subscriptionName(subscriptionName)
                       .subscriptionType(SubscriptionType.Shared)
                       .subscribe();
               System.out.println("consumer created");
   
               waitTillNextMinute();
   
               System.out.println("will unsub or close");
               Consumer<byte[]> ref = consumer;
               consumer.unsubscribeAsync().whenComplete((r, e) -> {
                   if (e == null) {
                       System.out.println("unsub succeeded");
                   } else {
                       e.printStackTrace();
                   }
                   ref.closeAsync().whenComplete((r2, e2) -> {
                       if (e2 == null) {
                           System.out.println("close succeeded");
                       } else {
                           e2.printStackTrace();
                       }
                   });
               });
   
               waitTillNextMinute();
           }
       }
   }
   
   ```
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services