----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/31706/#review75921 -----------------------------------------------------------
core/src/main/scala/kafka/consumer/PartitionAssignor.scala <https://reviews.apache.org/r/31706/#comment123248> Could we just import collection.mutable.HashMap since there seems no Java HashMap conflicts here? core/src/main/scala/kafka/consumer/PartitionAssignor.scala <https://reviews.apache.org/r/31706/#comment123250> Ditto. core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala <https://reviews.apache.org/r/31706/#comment123251> Ditto. core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala <https://reviews.apache.org/r/31706/#comment123253> Could we change it to "Partition assignor returns the global partition assignment organized as a map of [TopicPartition, ThreadId] per consumer, and we need to re-organize it to a map of [Partition, ThreadId] per topic before passing to the rebalance callback". core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123257> I am still a bit concerned about bounding ourselves with one producer per MM, becausing scaling MMs for the sake of scaling producer is kind of a waste of JVM containers / memory / etc. On the other hand, having one producer does give us the benefit of simple partition ordering semantics for the destination cluster. Maybe we can keep it as is but open for increasing the producer number if we bump into some cases that a single ioThread for sending is not sufficient. core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123259> There is no offset commit thread anymore right? core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123266> Capitalized "C" core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123267> Why add ";"? core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123268> Could you use !exitingOnSendFailure only in the outer loop, and in the onComplete callback, setting both boolean to true upon exception? core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123269> Is there a race condition, such that a thread could call producer.flush() while some other threads are also calling send() to the producer (in the current implementation flush does not block concurrent send), such that when it calls commitOffsets there are some offsets that get committed but not acked yet? core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123272> Is there another potential race condition, as concurrent threads could call send() while flush() is in progress? core/src/main/scala/kafka/tools/MirrorMaker.scala <https://reviews.apache.org/r/31706/#comment123274> Could we use Collection.singletonList here? - Guozhang Wang On March 10, 2015, 1:55 a.m., Jiangjie Qin wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/31706/ > ----------------------------------------------------------- > > (Updated March 10, 2015, 1:55 a.m.) > > > Review request for kafka. > > > Bugs: KAFKA-1997 > https://issues.apache.org/jira/browse/KAFKA-1997 > > > Repository: kafka > > > Description > ------- > > Addressed Guozhang's comments. > > > Changed the exit behavior on send failure because close(0) is not ready yet. > Will submit followup patch after KAFKA-1660 is checked in. > > > Expanded imports from _ and * to full class path > > > Incorporated Joel's comments. > > > Addressed Joel's comments. > > > Diffs > ----- > > > clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java > d5c79e2481d5e9a2524ac2ef6a6879f61cb7cb5f > core/src/main/scala/kafka/consumer/PartitionAssignor.scala > e6ff7683a0df4a7d221e949767e57c34703d5aad > core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala > 5487259751ebe19f137948249aa1fd2637d2deb4 > core/src/main/scala/kafka/javaapi/consumer/ConsumerRebalanceListener.java > 7f45a90ba6676290172b7da54c15ee5dc1a42a2e > core/src/main/scala/kafka/tools/MirrorMaker.scala > bafa379ff57bc46458ea8409406f5046dc9c973e > core/src/test/scala/unit/kafka/consumer/PartitionAssignorTest.scala > 543070f4fd3e96f3183cae9ee2ccbe843409ee58 > > core/src/test/scala/unit/kafka/consumer/ZookeeperConsumerConnectorTest.scala > 19640cc55b5baa0a26a808d708b7f4caf491c9f0 > > Diff: https://reviews.apache.org/r/31706/diff/ > > > Testing > ------- > > > Thanks, > > Jiangjie Qin > >
