You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2013/09/11 19:04:12 UTC
[27/36] git commit: kafka-649; patch v5; Cleanup log4j logging;
patched by Guozhang Wang; reviewed by Jun Rao
kafka-649; patch v5; Cleanup log4j logging; patched by Guozhang Wang; reviewed by Jun Rao
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/ceb55cad
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/ceb55cad
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/ceb55cad
Branch: refs/heads/trunk
Commit: ceb55cad5863fa7c39615e48a5d6358eb416a83e
Parents: 6849da0
Author: Guozhang Wang <gu...@linkedin.com>
Authored: Mon Aug 19 08:17:26 2013 -0700
Committer: Jun Rao <ju...@gmail.com>
Committed: Mon Aug 19 08:17:26 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/kafka/consumer/SimpleConsumer.scala | 2 +-
core/src/main/scala/kafka/server/AbstractFetcherThread.scala | 8 ++++----
core/src/main/scala/kafka/server/KafkaApis.scala | 3 +--
core/src/main/scala/kafka/server/KafkaRequestHandler.scala | 2 +-
core/src/main/scala/kafka/tools/MirrorMaker.scala | 5 +++--
5 files changed, 10 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/ceb55cad/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/SimpleConsumer.scala b/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
index 1c28328..4395fe3 100644
--- a/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
+++ b/core/src/main/scala/kafka/consumer/SimpleConsumer.scala
@@ -73,7 +73,7 @@ class SimpleConsumer(val host: String,
response = blockingChannel.receive()
} catch {
case e : java.io.IOException =>
- info("Reconnect due to socket error: ", e)
+ info("Reconnect due to socket error: %s".format(e.getMessage))
// retry once
try {
reconnect()
http://git-wip-us.apache.org/repos/asf/kafka/blob/ceb55cad/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
index 7663fac..d5addb3 100644
--- a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
@@ -144,17 +144,17 @@ abstract class AbstractFetcherThread(name: String, clientId: String, sourceBroke
try {
val newOffset = handleOffsetOutOfRange(topicAndPartition)
partitionMap.put(topicAndPartition, newOffset)
- warn("current offset %d for partition [%s,%d] out of range; reset offset to %d"
+ warn("Current offset %d for partition [%s,%d] out of range; reset offset to %d"
.format(currentOffset.get, topic, partitionId, newOffset))
} catch {
case e =>
- warn("error getting offset for partition [%s,%d] to broker %d".format(topic, partitionId, sourceBroker.id), e)
+ warn("Error getting offset for partition [%s,%d] to broker %d".format(topic, partitionId, sourceBroker.id), e)
partitionsWithError += topicAndPartition
}
case _ =>
if (isRunning.get) {
- warn("error for partition [%s,%d] to broker %d".format(topic, partitionId, sourceBroker.id),
- ErrorMapping.exceptionFor(partitionData.error))
+ warn("Error for partition [%s,%d] to broker %d:%s".format(topic, partitionId, sourceBroker.id,
+ ErrorMapping.exceptionFor(partitionData.error).getClass))
partitionsWithError += topicAndPartition
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/ceb55cad/core/src/main/scala/kafka/server/KafkaApis.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 208e3ef..b17964e 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -350,8 +350,7 @@ class KafkaApis(val requestChannel: RequestChannel,
BrokerTopicStats.getBrokerTopicStats(topic).failedFetchRequestRate.mark()
BrokerTopicStats.getBrokerAllTopicsStats.failedFetchRequestRate.mark()
error("Error when processing fetch request for partition [%s,%d] offset %d from %s with correlation id %d"
- .format(topic, partition, offset, if (isFetchFromFollower) "follower" else "consumer", fetchRequest.correlationId),
- t)
+ .format(topic, partition, offset, if (isFetchFromFollower) "follower" else "consumer", fetchRequest.correlationId), t)
new FetchResponsePartitionData(ErrorMapping.codeFor(t.getClass.asInstanceOf[Class[Throwable]]), -1L, MessageSet.Empty)
}
(TopicAndPartition(topic, partition), partitionData)
http://git-wip-us.apache.org/repos/asf/kafka/blob/ceb55cad/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
index fed0b86..4028118 100644
--- a/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
+++ b/core/src/main/scala/kafka/server/KafkaRequestHandler.scala
@@ -41,7 +41,7 @@ class KafkaRequestHandler(id: Int, brokerId: Int, val requestChannel: RequestCha
trace("Kafka request handler %d on broker %d handling request %s".format(id, brokerId, req))
apis.handle(req)
} catch {
- case e: Throwable => error("Exception when handling request")
+ case e: Throwable => error("Exception when handling request", e)
}
}
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/ceb55cad/core/src/main/scala/kafka/tools/MirrorMaker.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/MirrorMaker.scala b/core/src/main/scala/kafka/tools/MirrorMaker.scala
index c747bfb..6fb545a 100644
--- a/core/src/main/scala/kafka/tools/MirrorMaker.scala
+++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala
@@ -181,6 +181,7 @@ object MirrorMaker extends Logging {
private val shutdownLatch = new CountDownLatch(1)
private val threadName = "mirrormaker-" + threadId
+ this.logIdent = "[%s] ".format(threadName)
this.setName(threadName)
@@ -204,10 +205,10 @@ object MirrorMaker extends Logging {
}
} catch {
case e =>
- fatal("%s stream unexpectedly exited.", e)
+ fatal("Stream unexpectedly exited.", e)
} finally {
shutdownLatch.countDown()
- info("Stopped thread %s.".format(threadName))
+ info("Stopped thread.")
}
}