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/01/26 07:04:36 UTC
[5/28] git commit: KAFKA-707 Improve error message in the producer
when sending data to a partition without an active leader; reviewed by Jun Rao
KAFKA-707 Improve error message in the producer when sending data to a partition without an active leader; 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/426ef166
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/426ef166
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/426ef166
Branch: refs/heads/trunk
Commit: 426ef1662590893ff27a4be0b5f17b73115f154c
Parents: 2d05914
Author: Neha Narkhede <ne...@gmail.com>
Authored: Mon Jan 21 00:20:12 2013 -0800
Committer: Neha Narkhede <ne...@gmail.com>
Committed: Mon Jan 21 00:20:12 2013 -0800
----------------------------------------------------------------------
.../kafka/producer/async/DefaultEventHandler.scala | 3 ++-
1 files changed, 2 insertions(+), 1 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/426ef166/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
index 9a4e4bc..05e7c6c 100644
--- a/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
+++ b/core/src/main/scala/kafka/producer/async/DefaultEventHandler.scala
@@ -228,7 +228,8 @@ class DefaultEventHandler[K,V](config: ProducerConfig,
*/
private def send(brokerId: Int, messagesPerTopic: Map[TopicAndPartition, ByteBufferMessageSet]) = {
if(brokerId < 0) {
- warn("Failed to send to broker %d with data %s".format(brokerId, messagesPerTopic))
+ warn("Failed to send data %s since partitions %s don't have a leader".format(messagesPerTopic.map(_._2),
+ messagesPerTopic.map(_._1.toString).mkString(",")))
messagesPerTopic.keys.toSeq
} else if(messagesPerTopic.size > 0) {
val currentCorrelationId = correlationId.getAndIncrement