----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/51112/#review146580 -----------------------------------------------------------
I have concerns about this being safe. See especially my comments in KafkaChannel.java about ABA problems. I wonder if we could use dualCommitEnabled to support not only concurrent migration but also a rolling upgrade? Or do you think rolling upgrade of Flume agents simply cannot be supported? Sorry, I don't know Kafka very well. It's been a long time since I built it and I don't really know the internals very well. I just did some code spelunking yesterday and this morning. flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java (line 130) <https://reviews.apache.org/r/51112/#comment213082> nit: s/fro/from/ flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java (line 309) <https://reviews.apache.org/r/51112/#comment213089> style nit: return after this line and get rid of the "else" flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java (line 314) <https://reviews.apache.org/r/51112/#comment213090> style nit: return after this line and get rid of the following "else" flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java (line 320) <https://reviews.apache.org/r/51112/#comment213091> How does this interact with consumer groups and multiple Flume agents? Seems like this whole thing could be prone to ABA problems. It seems like migration is supported with a "dual commit" strategy. I wonder if it's usable in our context? See https://archive.cloudera.com/kafka/kafka/2/kafka-0.9.0-kafka2.0.1/implementation.html#offsetmigration Looks like you ultimately would have to use the ZookeeperConsumerConnector to take advantage of dual commit. Seems like you have to use Consumer.create() to get that behavior. Not sure exactly. flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java (line 322) <https://reviews.apache.org/r/51112/#comment213241> If we enforced that the offsets were all >= than the ones we committed, would this be safe under concurrency? flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java (line 337) <https://reviews.apache.org/r/51112/#comment213085> nit: Please use " : " style in the for-loop, here and below flume-ng-doc/sphinx/FlumeUserGuide.rst (line 2734) <https://reviews.apache.org/r/51112/#comment213083> This part needs a rebase (indentation change). Also, how about the following minor wording changes? When no Kafka stored offset is found, look up the offsets in Zookeeper and commit them to Kafka. This should be true to support seamless Kafka client migration from older versions of Flume. Once migrated this can be set to false, though that should generally not be required. If no Zookeeper offset is found the kafka.consumer.auto.offset.reset configuration defines how offsets are handled. - Mike Percy On Aug. 16, 2016, 12:30 p.m., Grant Henke wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/51112/ > ----------------------------------------------------------- > > (Updated Aug. 16, 2016, 12:30 p.m.) > > > Review request for Flume. > > > Bugs: FLUME-2972 > https://issues.apache.org/jira/browse/FLUME-2972 > > > Repository: flume-git > > > Description > ------- > > Offsets tracking the position in Kafka consumers change from using zookeeper > for offset storage to Kafka when moving from 0.8.x to 0.9.x. > FLUME-2823 makes the client change in the Kafka Channel but does not ensure > existing offsets get migrated in order to continue consuming where it left > off. > > This patch adds automated logic on startup to check if Kafka offsets exist, > if not and migration is enabled (by default) then the offsets from Zookeeper > are copied and committed to Kafka. > > > Diffs > ----- > > flume-ng-channels/flume-kafka-channel/pom.xml 587b4b4 > > flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannel.java > 90e3288 > > flume-ng-channels/flume-kafka-channel/src/main/java/org/apache/flume/channel/kafka/KafkaChannelConfiguration.java > ccf46d9 > > flume-ng-channels/flume-kafka-channel/src/test/java/org/apache/flume/channel/kafka/TestKafkaChannel.java > b63ac9b > flume-ng-doc/sphinx/FlumeUserGuide.rst fde9ff7 > > Diff: https://reviews.apache.org/r/51112/diff/ > > > Testing > ------- > > Unit tests so far. > > > Thanks, > > Grant Henke > >
