Repository: kafka Updated Branches: refs/heads/trunk 54e53b731 -> 8cff9119f
KAFKA-1890 Fix bug preventing Mirror Maker from successful rebalance; reviewed by Gwen Shapira and Neha Narkhede Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/8cff9119 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/8cff9119 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/8cff9119 Branch: refs/heads/trunk Commit: 8cff9119f8d125e55898e2095512b90826522336 Parents: 54e53b7 Author: Jiangjie Qin <[email protected]> Authored: Fri Jan 23 09:07:12 2015 -0800 Committer: Neha Narkhede <[email protected]> Committed: Fri Jan 23 09:07:26 2015 -0800 ---------------------------------------------------------------------- core/src/main/scala/kafka/tools/MirrorMaker.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/8cff9119/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 a11dbc8..3ee84ed 100644 --- a/core/src/main/scala/kafka/tools/MirrorMaker.scala +++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala @@ -213,11 +213,11 @@ object MirrorMaker extends Logging with KafkaMetricsGroup { val customRebalanceListenerClass = options.valueOf(consumerRebalanceListenerOpt) val customRebalanceListener = { if (customRebalanceListenerClass != null) - Utils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass) + Some(Utils.createObject[ConsumerRebalanceListener](customRebalanceListenerClass)) else - null + None } - consumerRebalanceListener = new InternalRebalanceListener(mirrorDataChannel, Some(customRebalanceListener)) + consumerRebalanceListener = new InternalRebalanceListener(mirrorDataChannel, customRebalanceListener) connector.setConsumerRebalanceListener(consumerRebalanceListener) // create producer threads
