You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by ch...@apache.org on 2016/01/06 20:00:20 UTC
incubator-twill git commit: (TWILL-151) Improve Logging error when
fetching message after Kafka server is stopped
Repository: incubator-twill
Updated Branches:
refs/heads/master b740da438 -> 323e5af4e
(TWILL-151) Improve Logging error when fetching message after Kafka server is stopped
This closes #65 on GitHub.
Signed-off-by: Terence Yim <ch...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-twill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-twill/commit/323e5af4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-twill/tree/323e5af4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-twill/diff/323e5af4
Branch: refs/heads/master
Commit: 323e5af4e536f3b5f306a5c59cd6787bf3a99818
Parents: b740da4
Author: sanojkodikkara <sa...@gmail.com>
Authored: Wed Sep 30 16:23:08 2015 +0530
Committer: Terence Yim <ch...@apache.org>
Committed: Wed Jan 6 10:55:36 2016 -0800
----------------------------------------------------------------------
.../internal/kafka/client/SimpleKafkaConsumer.java | 15 ++++++++++++---
1 file changed, 12 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/323e5af4/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java
index 8cfe889..0299e56 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java
@@ -28,6 +28,7 @@ import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Futures;
+
import kafka.api.FetchRequest;
import kafka.api.FetchRequestBuilder;
import kafka.api.PartitionOffsetRequestInfo;
@@ -39,6 +40,7 @@ import kafka.javaapi.OffsetResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.message.MessageAndOffset;
+
import org.apache.twill.common.Cancellable;
import org.apache.twill.common.Threads;
import org.apache.twill.kafka.client.BrokerInfo;
@@ -49,6 +51,7 @@ import org.apache.twill.kafka.client.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.net.ConnectException;
import java.nio.channels.ClosedByInterruptException;
import java.util.Iterator;
import java.util.List;
@@ -374,9 +377,15 @@ final class SimpleKafkaConsumer implements KafkaConsumer {
invokeCallback(messages, offset);
backoff.reset();
} catch (Throwable t) {
- if (running || !(t instanceof ClosedByInterruptException)) {
- // Only log if it is still running, otherwise, it just the interrupt caused by the stop.
- LOG.info("Exception when fetching message on {}.", topicPart, t);
+ // Only log if it is still running, otherwise, it just the interrupt caused by the stop.
+ if (!running) {
+ LOG.debug("Unable to fetch messages on {}, kafka consumer service shutdown is in progress.", topicPart);
+ } else {
+ if (t instanceof ClosedByInterruptException || t instanceof ConnectException) {
+ LOG.debug("Unable to fetch messages on {}, kafka server shutdown is in progress.", topicPart);
+ } else {
+ LOG.info("Exception when fetching message on {}.", topicPart, t);
+ }
backoff.backoff();
}
consumers.refresh(consumerEntry.getKey());