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/07/11 08:37:19 UTC
[09/40] git commit: KAFKA-916;
Break deadlock between fetcher shutdown and error partition handling;
reviewed by Jun Rao.
KAFKA-916; Break deadlock between fetcher shutdown and error partition handling; 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/517af477
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/517af477
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/517af477
Branch: refs/heads/trunk
Commit: 517af4779aed73efa9889b89d99a9c789f091b06
Parents: 59599cc
Author: Joel Koshy <jj...@gmail.com>
Authored: Tue May 28 09:55:39 2013 -0700
Committer: Joel Koshy <jj...@gmail.com>
Committed: Tue May 28 09:55:39 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala | 1 -
core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala | 1 +
2 files changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/517af477/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala b/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
index db104f1..96bd886 100644
--- a/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
+++ b/core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
@@ -167,7 +167,6 @@ class ConsumerFetcherManager(private val consumerIdString: String,
lock.lock()
try {
if (partitionMap != null) {
- partitionList.foreach(tp => removeFetcher(tp.topic, tp.partition))
noLeaderPartitionSet ++= partitionList
cond.signalAll()
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/517af477/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala b/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
index 1270e92..dda0a8f 100644
--- a/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
+++ b/core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
@@ -66,6 +66,7 @@ class ConsumerFetcherThread(name: String,
// any logic for partitions whose leader has changed
def handlePartitionsWithErrors(partitions: Iterable[TopicAndPartition]) {
+ partitions.foreach(tap => removePartition(tap.topic, tap.partition))
consumerFetcherManager.addPartitionsWithError(partitions)
}
}