[ISSUE apache#2152] Add isRunning method in DefaultLitePullConsumerImpl class and test suit (#2302)

client/src/main/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumer.java

@@ -211,6 +211,11 @@ public class DefaultLitePullConsumer extends ClientConfig implements LitePullCon
         this.defaultLitePullConsumerImpl.shutdown();
     }
 
+    @Override
+    public boolean isRunning() {
+        return this.defaultLitePullConsumerImpl.isRunning();
+    }
+
     @Override
     public void subscribe(String topic, String subExpression) throws MQClientException {
         this.defaultLitePullConsumerImpl.subscribe(withNamespace(topic), subExpression);

client/src/main/java/org/apache/rocketmq/client/consumer/LitePullConsumer.java

@@ -35,6 +35,13 @@ public interface LitePullConsumer {
      */
     void shutdown();
 
+    /**
+     * This consumer is still running
+     *
+     * @return true if consumer is still running
+     */
+    boolean isRunning();
+
     /**
      * Subscribe some topic with subExpression
      *

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java

@@ -229,6 +229,10 @@ public class DefaultLitePullConsumerImpl implements MQConsumerInner {
         }
     }
 
+    public synchronized boolean isRunning() {
+        return this.serviceState == ServiceState.RUNNING;
+    }
+
     public synchronized void start() throws MQClientException {
         switch (this.serviceState) {
             case CREATE_JUST:

client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java

@@ -513,6 +513,32 @@ public class DefaultLitePullConsumerTest {
         assertThat(offset).isEqualTo(100);
     }
 
+    @Test
+    public void testConsumerAfterShutdown() throws Exception {
+        DefaultLitePullConsumer defaultLitePullConsumer = createStartLitePullConsumer();
+        defaultLitePullConsumer.setNamesrvAddr("127.0.0.1:9876");
+        defaultLitePullConsumer.subscribe(topic, "*");
+        new AsyncConsumer().executeAsync(defaultLitePullConsumer);
+        Thread.sleep(10 * 1000);
+        defaultLitePullConsumer.shutdown();
+        assertThat(defaultLitePullConsumer.isRunning()).isFalse();
+    }
+
+    static class AsyncConsumer {
+        public void executeAsync(final DefaultLitePullConsumer consumer) {
+            new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    while (consumer.isRunning()) {
+                        List<MessageExt> poll = consumer.poll(2 * 1000);
+                        System.out.println("consumer is still running");
+                    }
+                    System.out.println("consumer shutdown");
+                }
+            }).start();
+        }
+    }
+
     private void initDefaultLitePullConsumer(DefaultLitePullConsumer litePullConsumer) throws Exception {
 
         Field field = DefaultLitePullConsumer.class.getDeclaredField("defaultLitePullConsumerImpl");