----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34524/#review85641 -----------------------------------------------------------
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java <https://reviews.apache.org/r/34524/#comment137304> Should we split this up into two checks? If the response is UNKNOWN_CONSUMER_ID, you might want to additionally reset the consumer id here. I was thinking something like this: ``` } else if (response.errorCode() == Errors.ILLEGAL_GENERATION.code()) { subscriptions.needReassignment(); } else if (response.errorCode() == Errors.UNKNOWN_CONSUMER_ID.code()) { this.consumerId = JoinGroupRequest.UNKNOWN_CONSUMER_ID; subscriptions.needReassignment(); } ``` core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34524/#comment137311> Let's say a consumer sends a JoinGroupRequest for a new group g and provides his own non-unknown consumer id (it could be a faulty implementation of the new consumer). The Coordinator would notice group == null, make the group, notice that the group doesn't contain the non-unknown consumer id, and then reply with UNKNOWN_CONSUMER_ID. So basically an empty, stable group has been made with no consumers. core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34524/#comment137318> Is returning NOT_COORDINATOR_FOR_CONSUMER right? By this point in handleJoinGroup, we've already verified that we are the coordinator for the group. core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala <https://reviews.apache.org/r/34524/#comment137317> Is returning NOT_COORDINATOR_FOR_CONSUMER right? By this point in handleHeartbeat, we've already verified that we are the coordinator for the group. - Onur Karaman On May 21, 2015, 2:15 a.m., Guozhang Wang wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/34524/ > ----------------------------------------------------------- > > (Updated May 21, 2015, 2:15 a.m.) > > > Review request for kafka. > > > Bugs: KAFKA-2208 > https://issues.apache.org/jira/browse/KAFKA-2208 > > > Repository: kafka > > > Description > ------- > > 1. Add error handling on consumer; 2. Add the max / min consumer session > timeout to kafka server configs; 3. Fixed some consumer bouncing tests > > > Diffs > ----- > > > clients/src/main/java/org/apache/kafka/clients/consumer/internals/Coordinator.java > b2764df11afa7a99fce46d1ff48960d889032d14 > > clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java > ef9dd5238fbc771496029866ece1d85db6d7b7a5 > > clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java > f548cd0ef70929b35ac887f8fccb7b24c3e2c11a > > clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java > fd9c545c99058ad3fbe3b2c55ea8b6ea002f5a51 > core/src/main/scala/kafka/coordinator/ConsumerCoordinator.scala > af06ad45cdc46ac3bc27898ebc1a5bd5b1c7b19e > core/src/main/scala/kafka/coordinator/ConsumerGroupMetadata.scala > 47bdfa7cc86fd4e841e2b1d6bfd40f1508e643bd > core/src/main/scala/kafka/network/RequestChannel.scala > 1d0024c8f0c2ab0efa6d8cfca6455877a6ed8026 > core/src/main/scala/kafka/server/KafkaConfig.scala > 9efa15ca5567b295ab412ee9eea7c03eb4cdc18b > core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala > 5c4cca653b3801df3494003cc40a56ae60a789a6 > core/src/test/scala/integration/kafka/api/ConsumerTest.scala > a1eed965a148eb19d9a6cefbfce131f58aaffc24 > core/src/test/scala/unit/kafka/server/KafkaConfigConfigDefTest.scala > 8014a5a6c362785539f24eb03d77278434614fe6 > > Diff: https://reviews.apache.org/r/34524/diff/ > > > Testing > ------- > > > Thanks, > > Guozhang Wang > >
