----------------------------------------------------------- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/17248/#review35293 -----------------------------------------------------------
Sorry for the late review. The high level comment is that we need to make this change backward compatible, which means that we can do a rolling upgrade of an existing 0.8 cluster. To do that, we need to (1) consider wire protocol changes (see comments on Broker); (2) make sure that the old code can read the new ZK format. Once the following comments are addressed, could you exercise a rolling upgrade and see if there is any issue? core/src/main/scala/kafka/admin/AdminUtils.scala <https://reviews.apache.org/r/17248/#comment65764> not => Not core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala <https://reviews.apache.org/r/17248/#comment65766> Shouldn't we pass in maxReplicaPerRack? core/src/main/scala/kafka/admin/TopicCommand.scala <https://reviews.apache.org/r/17248/#comment65768> After you rebase, we need to make sure this option is only available during topic creation. core/src/main/scala/kafka/client/ClientUtils.scala <https://reviews.apache.org/r/17248/#comment65760> broker list is for the clients and there is no need for them to know the rack a broker is in. host/port is enough. core/src/main/scala/kafka/cluster/Broker.scala <https://reviews.apache.org/r/17248/#comment65770> Could we preserve the current constructor w/o rack and default rack to 0? core/src/main/scala/kafka/cluster/Broker.scala <https://reviews.apache.org/r/17248/#comment65763> This actually changes the wire protocol btw the controller and the broker, which makes rolling upgrade difficult. If we really want to change the protocol, we need to do this in 2 steps. First upgrade the code w/o using the new format. Second, use the new format. So, we will need a new config option. For this particular case, I am not sure that we need to change the protocol though. We can only populate the rack info when instantiating a broker from ZK. This will make sure that we have the rack info for replica assignment. Then, we don't change the wire protocol btw the controller and the broker. All brokers instantiated through ByteBuffer will have the default rack info. core/src/main/scala/kafka/utils/ZkUtils.scala <https://reviews.apache.org/r/17248/#comment65769> Could this be integrated into getCluster()? - Jun Rao On Feb. 1, 2014, 12:20 a.m., Joris Van Remoortere wrote: > > ----------------------------------------------------------- > This is an automatically generated e-mail. To reply, visit: > https://reviews.apache.org/r/17248/ > ----------------------------------------------------------- > > (Updated Feb. 1, 2014, 12:20 a.m.) > > > Review request for kafka. > > > Bugs: KAFKA-1215 > https://issues.apache.org/jira/browse/KAFKA-1215 > > > Repository: kafka > > > Description > ------- > > KAFKA-1226 > > > Diffs > ----- > > core/src/main/scala/kafka/admin/AdminUtils.scala a167756 > core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala 2637586 > core/src/main/scala/kafka/admin/TopicCommand.scala 842c110 > core/src/main/scala/kafka/client/ClientUtils.scala 1d2f81b > core/src/main/scala/kafka/cluster/Broker.scala 9407ed2 > core/src/main/scala/kafka/controller/KafkaController.scala a0267ae > core/src/main/scala/kafka/controller/PartitionStateMachine.scala ac4262a > core/src/main/scala/kafka/server/KafkaApis.scala 29abc46 > core/src/main/scala/kafka/server/KafkaConfig.scala 3c3aafc > core/src/main/scala/kafka/server/KafkaHealthcheck.scala 9dca55c > core/src/main/scala/kafka/server/KafkaServer.scala 5e34f95 > core/src/main/scala/kafka/utils/ZkUtils.scala b42e52b > core/src/test/scala/unit/kafka/admin/AddPartitionsTest.scala 115e203 > core/src/test/scala/unit/kafka/admin/AdminTest.scala 59de1b4 > core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala > 8df0982 > core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala 9347ea6 > core/src/test/scala/unit/kafka/integration/FetcherTest.scala 47130d3 > core/src/test/scala/unit/kafka/integration/TopicMetadataTest.scala 9998a11 > core/src/test/scala/unit/kafka/producer/AsyncProducerTest.scala 18e3555 > core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala 38e3ae7 > core/src/test/scala/unit/kafka/utils/TestUtils.scala d88b6c3 > examples/README d33f6c5 > > Diff: https://reviews.apache.org/r/17248/diff/ > > > Testing > ------- > > > File Attachments > ---------------- > > rack_aware_replica_assignment_v1.patch > > https://reviews.apache.org/media/uploaded/files/2014/01/23/394cef99-f800-4d94-bc59-fdb6c68b53f5__rack_aware_replica_assignment_v1.patch > > > Thanks, > > Joris Van Remoortere > >
