(feat) unsubscribe topics before close

Author: ylgrgyq

src/main/java/cn/leancloud/kafka/consumer/Fetcher.java

@@ -138,6 +138,7 @@ private void gracefulShutdown() {
         final long start = System.currentTimeMillis();
         long remain = gracefulShutdownMillis;
         try {
+            consumer.unsubscribe();
             for (Future<ConsumerRecord<K, V>> future : pendingFutures.values()) {
                 try {
                     if (remain > 0) {
@@ -153,6 +154,7 @@ private void gracefulShutdown() {
             processCompletedRecords();
         } catch (InterruptedException ex) {
             logger.warn("Graceful shutdown was interrupted.");
+            Thread.currentThread().interrupt();
         } catch (ExecutionException ex) {
             logger.error("Handle message got unexpected exception. Continue shutdown without wait handling message done.", ex);
         }

src/test/java/cn/leancloud/kafka/consumer/FetcherTest.java

@@ -75,6 +75,7 @@ public void testGracefulShutdown() throws Exception {
         fetcherThread.join();
         assertThat(fetcher.pendingFutures()).isEmpty();
         verify(policy, times(1)).partialCommit();
+        assertThat(consumer.subscription()).isEmpty();
     }
 
     @Test