From c0fdcd305c8487610b7bdb2ac080b4a1055534f3 Mon Sep 17 00:00:00 2001 From: Jun Rao Date: Fri, 12 Oct 2012 20:28:14 +0000 Subject: [PATCH] Replication Data Loss in Mirror Maker Bouncing testcase; patched by Jun Rao; reviewed by Joel Koshy, John Fung and Neha Narkhede; kafka-567 git-svn-id: https://svn.apache.org/repos/asf/incubator/kafka/branches/0.8@1397717 13f79535-47bb-0310-9956-ffa450edef68 --- .../main/scala/kafka/producer/async/ProducerSendThread.scala | 2 ++ 1 file changed, 2 insertions(+) diff --git a/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala b/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala index 25d995b5782..b0c3952ff90 100644 --- a/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala +++ b/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala @@ -93,6 +93,8 @@ class ProducerSendThread[K,V](val threadName: String, events = new ListBuffer[ProducerData[K,V]] } } + // send the last batch of events + tryToHandle(events) if(queue.size > 0) throw new IllegalQueueStateException("Invalid queue state! After queue shutdown, %d remaining items in the queue" .format(queue.size))