> On Dec. 18, 2014, 10:42 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/tools/MirrorMaker.scala, line 614 > > <https://reviews.apache.org/r/25995/diff/14/?file=795052#file795052line614> > > > > Should this be fatal? i.e., fatal is normally used before exiting > > (abnormally). WARN would be more suitable. > > > > I don't think it makes sense to "not advance" the offset here > > especially if you will still keep sending messages. I think you need to > > remove it from the unacked offset list. E.g., you may configure your mirror > > maker producer to only few retries (in which case you are okay with data > > loss). In this scenario you should just let the error go and allow the > > mirror maker to proceed normally. > > > > If someone wants zero data loss the MM should be configured with > > required acks -1 and infinite retries. > > > > Maybe I'm misunderstanding what zero data loss really means - can you > > clarify? (Especially if someone configures the producer with acks (say) one > > and limited retries) > > Jiangjie Qin wrote: > That makes sense. So I've changed the code to work in the following way: > > 1. If retries is set to infinite, the producer will keep retrying and the > entire pipeline will finally be blocked. (This is strict data-loss free.) > 2. If retries are not set to infinite, after the retries are exhausted, > it will remove the offset from unacked list and record it as a > skippedUnackedMessage, which is an exposed metric.
For (1) will the pipeline be blocked though? i.e., the consumer will continue to send messages to the data channel for all partitions. Also, we will be susceptible to reordering for the partition that is being retried even if maxInFlightRequests is set to one. (check wiki) It would be useful to very clearly outline the guarantees that the mirror maker provides and under what configurations. e.g., zero data loss, whether there will be duplicates, whether reordering is possible. Also, it may make sense to add a --lossless flag to the mirror maker that automatically ensures the configs are set correctly: * acks to -1 (or all) * retries to infinity * use new producer * block on buffer exhaustion * consumer autocommit off * anything else that may be needed This makes it much easier to document and use. What do you think? > On Dec. 18, 2014, 10:42 a.m., Joel Koshy wrote: > > core/src/main/scala/kafka/tools/MirrorMaker.scala, line 668 > > <https://reviews.apache.org/r/25995/diff/14/?file=795052#file795052line668> > > > > Should we make this a generic DoublyLinkedList data structure in utils > > or some other suitable place and unit test it as well? > > Jiangjie Qin wrote: > I'm not sure if this is generic enough to put it into utils. This raw > linked list seems only serve the purpose of removing/inserting node in the > middle in O(1), which cannot be achieved in java linkedlist. Maybe we can > keep it here now. And if later on there are some other use cases, we can > refactor the code to create a raw LinkedList in utils and use that one. What > do you think? Sort of agree, but only to some degree. The main issue is that you cannot unit test this - unless we come up with "mirror maker" unit tests that have 100 percent code coverage of this utility. That's why it's generally a sign that the utility needs to be moved out so that it can be tested thoroughly. http://stackoverflow.com/questions/3353318/how-do-i-test-local-inner-class-methods-in-java - Joel ----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/25995/#review65477 ----------------------------------------------------------- On Dec. 19, 2014, 7:41 a.m., Jiangjie Qin wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/25995/ > ----------------------------------------------------------- > > (Updated Dec. 19, 2014, 7:41 a.m.) > > > Review request for kafka. > > > Bugs: KAFKA-1650 and KAKFA-1650 > https://issues.apache.org/jira/browse/KAFKA-1650 > https://issues.apache.org/jira/browse/KAKFA-1650 > > > Repository: kafka > > > Description > ------- > > Addressed Guozhang's comments. > > > Addressed Guozhang's comments > > > commit before switch to trunk > > > commit before rebase > > > Rebased on trunk, Addressed Guozhang's comments. > > > Addressed Guozhang's comments on MaxInFlightRequests > > > Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into > mirrormaker-redesign > > > Incorporated Guozhang's comments > > > Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into > mirrormaker-redesign > > > Merged KAFKA-345 into this patch. Incorporated Joel and Jun's comments. > > > Added consumer rebalance listener to mirror maker, will test it later. > > > Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into > mirrormaker-redesign > > Conflicts: > core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala > > core/src/test/scala/unit/kafka/consumer/ZookeeperConsumerConnectorTest.scala > > added custom config for consumer rebalance listener > > > Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into > mirrormaker-redesign > > > Add configurable consumer rebalance listener > > > Incorporated Guozhang's comments > > > Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into > mirrormaker-redesign > > > Incorporated Guozhang's comments. > > > Addressed Guozhang's comment. > > > numMessageUnacked should be decremented no matter the send was successful or > not. > > > Addressed Jun's comments. > > > Incorporated Jun's comments > > > Incorporated Jun's comments and rebased on trunk > > > Rebased on current trunk > > > Addressed Joel's comments. > > > Addressed Joel's comments. > > > Incorporated Joel's comments > > > Incorporated Joel's comments > > > Merge branch 'trunk' of http://git-wip-us.apache.org/repos/asf/kafka into > mirrormaker-redesign > > > Incorporated Joel's comments > > > Fix a bug in metric. > > > Missed some change in the prvevious patch submission, submit patch again. > > > change offset commit thread to use scheduler. > > > Diffs > ----- > > core/src/main/scala/kafka/consumer/ConsumerConnector.scala > 62c0686e816d2888772d5a911becf625eedee397 > core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala > e991d2187d03241f639eeaf6769fb59c8c99664c > core/src/main/scala/kafka/javaapi/consumer/ZookeeperConsumerConnector.scala > 9baad34a9793e5067d11289ece2154ba87b388af > core/src/main/scala/kafka/tools/MirrorMaker.scala > 53cb16c2949e0ac36a0e943564fc9fc9b4c84caa > > Diff: https://reviews.apache.org/r/25995/diff/ > > > Testing > ------- > > > Thanks, > > Jiangjie Qin > >
