Merge pull request #11 from leancloud/fix/fetcher-self-close

do not triger wake when fetch exit by itself

Author: ylgrgyq

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

@@ -144,15 +144,15 @@ public void run() {
                 }
             } catch (ExecutionException ex) {
                 unsubscribedStatus = UnsubscribedStatus.ERROR;
-                close();
+                markClosed();
                 break;
             } catch (Exception ex) {
                 if (ex instanceof InterruptedException) {
                     Thread.currentThread().interrupt();
                 }
 
                 unsubscribedStatus = UnsubscribedStatus.ERROR;
-                close();
+                markClosed();
                 logger.error("Fetcher quit with unexpected exception. Will rebalance after poll timeout.", ex);
                 break;
             }
@@ -164,7 +164,7 @@ public void run() {
 
     @Override
     public void close() {
-        closed = true;
+        markClosed();
         consumer.wakeup();
     }
 
@@ -177,6 +177,10 @@ Map<ConsumerRecord<K, V>, Future<ConsumerRecord<K, V>>> pendingFutures() {
         return pendingFutures;
     }
 
+    private void markClosed() {
+        closed = true;
+    }
+
     private boolean closed() {
         return closed;
     }

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

@@ -203,15 +203,15 @@ private void ensureInInit() {
 
     private String fetcherThreadName(Collection<String> topics) {
         final String firstTopic = topics.iterator().next();
-        String postfix = firstTopic.substring(0, min(10, firstTopic.length()));
-        postfix += topics.size() > 1 || firstTopic.length() > 10 ? "..." : "";
+        String postfix = firstTopic.substring(0, min(50, firstTopic.length()));
+        postfix += topics.size() > 1 || firstTopic.length() > 50 ? "..." : "";
         return "kafka-fetcher-for-" + postfix;
     }
 
     private String fetcherThreadName(Pattern pattern) {
         final String patternInString = pattern.toString();
-        String postfix = patternInString.substring(0, min(10, patternInString.length()));
-        postfix += patternInString.length() > 10 ? "..." : "";
+        String postfix = patternInString.substring(0, min(50, patternInString.length()));
+        postfix += patternInString.length() > 50 ? "..." : "";
         return "kafka-fetcher-for-" + postfix;
     }
 }