----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34450/#review84604 -----------------------------------------------------------
Thanks for the patch. A few comments below. core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34450/#comment135934> Another way to do this is to only load from ZK on the becoming leader event for an offsetTopic partition. Then, we don't have to read from ZK during join group, which will introduce unnecessary overhead when joining a new group. core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34450/#comment135970> I was thinking whether it's worth including the leader epoch (of the corresponding offset topic partition) in the ZK value as we did for leaderAndIsr to prevent a zombie consumer coordinator from overwriting the value, during a soft failure. I am not sure if it's worth doing this immediately because 1. When this happens, consumers can still recover after the heartbeat fails. 2. It seems that doing this right is a bit more complicated. We need to keep the leader epoch in the ZK value. However, during a leader change, we probably need to update the values in ZK with the new leader epoch as well, in order to truely prevent the zombie coordinator from overwriting the value. So, I think for now, we can just use the simple approach in this patch. core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34450/#comment135961> Should we encode topics as an array of String? core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34450/#comment135956> updatePersistentPath already handles the case when the node doesn't exist. core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34450/#comment135959> Not sure if we need to wrap unexpected exceptions. The callers already handles unexpected exceptions. core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34450/#comment135963> Should we auto fix this or throw a KafkaException here? If we do auto fix this, we probably should log a warning since this is not expected. Ditto for the handling of partitionStrategy. core/src/main/scala/kafka/coordinator/CoordinatorMetadata.scala <https://reviews.apache.org/r/34450/#comment135965> Could we just do groups(group.groupId)? core/src/main/scala/kafka/utils/ZkUtils.scala <https://reviews.apache.org/r/34450/#comment135967> This doesn't seem to be used. - Jun Rao On May 20, 2015, 4:13 p.m., Guozhang Wang wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/34450/ > ----------------------------------------------------------- > > (Updated May 20, 2015, 4:13 p.m.) > > > Review request for kafka. > > > Bugs: KAFKA-2017 > https://issues.apache.org/jira/browse/KAFKA-2017 > > > Repository: kafka > > > Description > ------- > > 1. Upon receiving join-group, if the group metadata cannot be found in the > local cache try to read it from ZK; 2. Upon completing rebalance, update the > ZK with new group registry or delete the registry if the group becomes empty > > > Diffs > ----- > > core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala > af06ad45cdc46ac3bc27898ebc1a5bd5b1c7b19e > core/src/main/scala/kafka/coordinator/ConsumerGroupMetadata.scala > 47bdfa7cc86fd4e841e2b1d6bfd40f1508e643bd > core/src/main/scala/kafka/coordinator/CoordinatorMetadata.scala > c39e6de34ee531c6dfa9107b830752bd7f8fbe59 > core/src/main/scala/kafka/utils/ZkUtils.scala > 2618dd39b925b979ad6e4c0abd5c6eaafb3db5d5 > > Diff: https://reviews.apache.org/r/34450/diff/ > > > Testing > ------- > > > Thanks, > > Guozhang Wang > >
