(fix) complete unsubscribe future even on close failed

Author: ylgrgyq

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

@@ -153,7 +153,7 @@ public void run() {
 
                 unsubscribedStatus = UnsubscribedStatus.ERROR;
                 markClosed();
-                logger.error("Fetcher quit with unexpected exception. Will rebalance after poll timeout.", ex);
+                logger.error("Fetcher quit with unexpected exception.", ex);
                 break;
             }
         }

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

@@ -189,8 +189,11 @@ private CompletableFuture<UnsubscribedStatus> setupUnsubscribedFuture() {
         // returned CompletableFuture
         final CompletableFuture<UnsubscribedStatus> ret = new CompletableFuture<>();
         fetcher.unsubscribeStatusFuture().thenAccept(status -> {
-            close();
-            ret.complete(status);
+            try {
+                close();
+            } finally {
+                ret.complete(status);
+            }
         });
         return ret;
     }