You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Jun Rao <ju...@gmail.com> on 2014/07/15 20:34:20 UTC
Review Request 23516: Patch for KAFKA-1462
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/
-----------------------------------------------------------
Review request for kafka.
Bugs: KAFKA-1462
https://issues.apache.org/jira/browse/KAFKA-1462
Repository: kafka
Description
-------
remove partition from all PartitionData since it's redundant
minor fixes
initial patch
Diffs
-----
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
Diff: https://reviews.apache.org/r/23516/diff/
Testing
-------
Thanks,
Jun Rao
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Jay Kreps <bo...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/#review48092
-----------------------------------------------------------
Ship it!
Ship It!
- Jay Kreps
On July 17, 2014, 4:39 a.m., Jun Rao wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/23516/
> -----------------------------------------------------------
>
> (Updated July 17, 2014, 4:39 a.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1462
> https://issues.apache.org/jira/browse/KAFKA-1462
>
>
> Repository: kafka
>
>
> Description
> -------
>
> address Jay's comments
>
>
> remove partition from all PartitionData since it's redundant
>
>
> minor fixes
>
>
> initial patch
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
> clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
> clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
> clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
> clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
> clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
> clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
> clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
> clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
> clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java PRE-CREATION
> clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
> clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
> core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
> core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
> core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
> core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
> core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
> core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
> core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
> core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
> core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
> core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
> core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
> core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
> core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
> core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
> core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
> core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
> core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
> core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
> core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
>
> Diff: https://reviews.apache.org/r/23516/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Jun Rao
>
>
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Jun Rao <ju...@gmail.com>.
> On July 17, 2014, 9:52 p.m., Guozhang Wang wrote:
> > I am not sure why we need to remove correlationId from the ReuqestOrResponse in order to do ser/deser. Shouldn't we just add correlation_id to the AbstractRequestResponse java class?
This is just to make reusing the client request objects easier. CorrelationId is already included in the header and digging it out to pass to the base constructor is inconvenient. There is also no clear usage of correlationId in the base class.
> On July 17, 2014, 9:52 p.m., Guozhang Wang wrote:
> > core/src/main/scala/kafka/controller/ControllerChannelManager.scala, line 138
> > <https://reviews.apache.org/r/23516/diff/2/?file=634219#file634219line138>
> >
> > Should we use toString or describe(false) here?
describe(false) is only useful to reduce log volume in TRACE. This is an error logging and a full description of the request will be useful.
> On July 17, 2014, 9:52 p.m., Guozhang Wang wrote:
> > core/src/main/scala/kafka/controller/ControllerChannelManager.scala, line 157
> > <https://reviews.apache.org/r/23516/diff/2/?file=634219#file634219line157>
> >
> > Ditto above.
Ditto as the above comment.
- Jun
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/#review48071
-----------------------------------------------------------
On July 17, 2014, 4:39 a.m., Jun Rao wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/23516/
> -----------------------------------------------------------
>
> (Updated July 17, 2014, 4:39 a.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1462
> https://issues.apache.org/jira/browse/KAFKA-1462
>
>
> Repository: kafka
>
>
> Description
> -------
>
> address Jay's comments
>
>
> remove partition from all PartitionData since it's redundant
>
>
> minor fixes
>
>
> initial patch
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
> clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
> clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
> clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
> clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
> clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
> clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
> clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
> clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
> clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java PRE-CREATION
> clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
> clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
> core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
> core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
> core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
> core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
> core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
> core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
> core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
> core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
> core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
> core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
> core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
> core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
> core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
> core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
> core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
> core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
> core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
> core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
> core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
>
> Diff: https://reviews.apache.org/r/23516/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Jun Rao
>
>
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Guozhang Wang <gu...@linkedin.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/#review48071
-----------------------------------------------------------
I am not sure why we need to remove correlationId from the ReuqestOrResponse in order to do ser/deser. Shouldn't we just add correlation_id to the AbstractRequestResponse java class?
core/src/main/scala/kafka/controller/ControllerChannelManager.scala
<https://reviews.apache.org/r/23516/#comment84348>
Should we use toString or describe(false) here?
core/src/main/scala/kafka/controller/ControllerChannelManager.scala
<https://reviews.apache.org/r/23516/#comment84349>
Ditto above.
- Guozhang Wang
On July 17, 2014, 4:39 a.m., Jun Rao wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/23516/
> -----------------------------------------------------------
>
> (Updated July 17, 2014, 4:39 a.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1462
> https://issues.apache.org/jira/browse/KAFKA-1462
>
>
> Repository: kafka
>
>
> Description
> -------
>
> address Jay's comments
>
>
> remove partition from all PartitionData since it's redundant
>
>
> minor fixes
>
>
> initial patch
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
> clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
> clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
> clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
> clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
> clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
> clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
> clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
> clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
> clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java PRE-CREATION
> clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
> clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
> core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
> core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
> core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
> core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
> core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
> core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
> core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
> core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
> core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
> core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
> core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
> core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
> core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
> core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
> core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
> core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
> core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
> core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
> core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
>
> Diff: https://reviews.apache.org/r/23516/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Jun Rao
>
>
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Jun Rao <ju...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/
-----------------------------------------------------------
(Updated July 17, 2014, 4:39 a.m.)
Review request for kafka.
Bugs: KAFKA-1462
https://issues.apache.org/jira/browse/KAFKA-1462
Repository: kafka
Description (updated)
-------
address Jay's comments
remove partition from all PartitionData since it's redundant
minor fixes
initial patch
Diffs (updated)
-----
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java PRE-CREATION
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
Diff: https://reviews.apache.org/r/23516/diff/
Testing
-------
Thanks,
Jun Rao
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Jay Kreps <bo...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/#review47954
-----------------------------------------------------------
clients/src/main/java/org/apache/kafka/common/Cluster.java
<https://reviews.apache.org/r/23516/#comment84197>
This should ideally be
List<PartitionInfo> partitionsForTopic(String topic)
clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
<https://reviews.apache.org/r/23516/#comment84198>
If we override equals we need to also override hashCode. Also think if we need to add a test to ProtocolSerializationTest
clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java
<https://reviews.apache.org/r/23516/#comment84199>
Can we come up with a better name? Structs are already generic. I think this is something like AbstractRequestResponse?
clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java
<https://reviews.apache.org/r/23516/#comment84201>
These group by methods don't belong in a generic base class.
clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java
<https://reviews.apache.org/r/23516/#comment84200>
hashCode
- Jay Kreps
On July 15, 2014, 6:36 p.m., Jun Rao wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/23516/
> -----------------------------------------------------------
>
> (Updated July 15, 2014, 6:36 p.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1462
> https://issues.apache.org/jira/browse/KAFKA-1462
>
>
> Repository: kafka
>
>
> Description
> -------
>
> 1. I kept the request objects in the server separated from those in the client. This is because (1) some of the existing request objects are part of old client api (FetechRequest, OffsetCommitRequest, etc) and we can't remove them until the old clients are removed, (2) changing existing request objects on the server side requires significant refactoring.
>
> 2. On the client side, I refactored existing request objects a bit. Now, every request/response object extends from a GenericStruct. GenericStruct provides a standard way for doing serialization and toString so that we don't have to do that on every request. Each request/response can be constructed in two ways: (1) by providing request specific fields; (2) by providing a struct. The latter is used for getting a request/response from its serialized format.
>
> 3. On the server side. What I did is to keep the existing requests more or less untouched. For new types of requests, create a thin wrapper on the server side so that it can leverage the request objects created on the client side. This way the server side object will share the serialization and the toString logic with the client side object. In order to do this, I removed correlationId from the RequestOrResponse object. There is only one place where correlationId is directly referenced and it is not necessary.
>
> 4. Multi-version support. We now need to support two versions of OffsetCommitRequest since for the new consumer work, we added two extra fields in the request. For simplicity, both versions are converted to the same request object. Since the old version doesn't have the new fields, defaults will be used.
>
> 5. The new requests/responses are based on the format described in https://cwiki.apache.org/confluence/display/KAFKA/Kafka+0.9+Consumer+Rewrite+Design#Kafka0.9ConsumerRewriteDesign-Requestformats. I made some minor changes to the wiki so that the new requests follow the current standard.
>
> 6. Added some missing util functions and added unit test for testing the serialization/deserialization logic.
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
> clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
> clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
> clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
> clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
> clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
> clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
> clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
> clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
> clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
> core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
> core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
> core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
> core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
> core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
> core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
> core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
> core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
> core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
> core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
> core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
> core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
> core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
> core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
> core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
> core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
> core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
> core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
> core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
>
> Diff: https://reviews.apache.org/r/23516/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Jun Rao
>
>
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Jun Rao <ju...@gmail.com>.
> On July 16, 2014, 11:09 p.m., Guozhang Wang wrote:
> > clients/src/main/java/org/apache/kafka/common/Cluster.java, line 18
> > <https://reviews.apache.org/r/23516/diff/1/?file=632640#file632640line18>
> >
> > Do we ever want to use * in imports?
IDE did the optimization since there are too many imports from the same package.
> On July 16, 2014, 11:09 p.m., Guozhang Wang wrote:
> > clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java, line 117
> > <https://reviews.apache.org/r/23516/diff/1/?file=632642#file632642line117>
> >
> > What is this field used for?
This is to stick any user specific metadata (e.g., external offset in an indexer that needs to match the Kafka offset).
> On July 16, 2014, 11:09 p.m., Guozhang Wang wrote:
> > clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java, line 177
> > <https://reviews.apache.org/r/23516/diff/1/?file=632642#file632642line177>
> >
> > I thought the broker will also check consumer id and generation id for offset fetch request?
Those fields are needed in OffsetCommitRequests since they are changing the offset values. Not sure if we need to do the check for the reader though.
> On July 16, 2014, 11:09 p.m., Guozhang Wang wrote:
> > clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java, line 224
> > <https://reviews.apache.org/r/23516/diff/1/?file=632642#file632642line224>
> >
> > This is a long lasting problem: replica_id here can not only be the id of a follower replica, but also one of the two values: "-1" for ordinary consumer, or "-2" for debugging consumer. We use this field into folds, 1) logging for trouble shooting, which can be helpful only when this is from a follower replica, 2) deciding if it is from the consumer or a replica, for this purpose we do not really care about this id value. We should probably rename this field.
We can do that, but that needs to be done consistently in other places. Could you file a separate jira?
> On July 16, 2014, 11:09 p.m., Guozhang Wang wrote:
> > clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java, line 328
> > <https://reviews.apache.org/r/23516/diff/1/?file=632642#file632642line328>
> >
> > For the customized rebalancer, will that be supported as: 1) user implement their rebalancer and re-deploy the server with this new code, 2) bounce the consumer setting the new rebalancer? Here I assume the "strategy" will just be the class name?
We will probably have a few predefined strategies. Adding customized strategies requires a broker bounce. Not sure if we will use the class name or some numerical identifier.
- Jun
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/#review47940
-----------------------------------------------------------
On July 15, 2014, 6:36 p.m., Jun Rao wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/23516/
> -----------------------------------------------------------
>
> (Updated July 15, 2014, 6:36 p.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1462
> https://issues.apache.org/jira/browse/KAFKA-1462
>
>
> Repository: kafka
>
>
> Description
> -------
>
> 1. I kept the request objects in the server separated from those in the client. This is because (1) some of the existing request objects are part of old client api (FetechRequest, OffsetCommitRequest, etc) and we can't remove them until the old clients are removed, (2) changing existing request objects on the server side requires significant refactoring.
>
> 2. On the client side, I refactored existing request objects a bit. Now, every request/response object extends from a GenericStruct. GenericStruct provides a standard way for doing serialization and toString so that we don't have to do that on every request. Each request/response can be constructed in two ways: (1) by providing request specific fields; (2) by providing a struct. The latter is used for getting a request/response from its serialized format.
>
> 3. On the server side. What I did is to keep the existing requests more or less untouched. For new types of requests, create a thin wrapper on the server side so that it can leverage the request objects created on the client side. This way the server side object will share the serialization and the toString logic with the client side object. In order to do this, I removed correlationId from the RequestOrResponse object. There is only one place where correlationId is directly referenced and it is not necessary.
>
> 4. Multi-version support. We now need to support two versions of OffsetCommitRequest since for the new consumer work, we added two extra fields in the request. For simplicity, both versions are converted to the same request object. Since the old version doesn't have the new fields, defaults will be used.
>
> 5. The new requests/responses are based on the format described in https://cwiki.apache.org/confluence/display/KAFKA/Kafka+0.9+Consumer+Rewrite+Design#Kafka0.9ConsumerRewriteDesign-Requestformats. I made some minor changes to the wiki so that the new requests follow the current standard.
>
> 6. Added some missing util functions and added unit test for testing the serialization/deserialization logic.
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
> clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
> clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
> clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
> clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
> clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
> clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
> clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
> clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
> clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
> core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
> core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
> core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
> core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
> core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
> core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
> core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
> core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
> core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
> core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
> core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
> core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
> core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
> core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
> core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
> core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
> core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
> core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
> core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
>
> Diff: https://reviews.apache.org/r/23516/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Jun Rao
>
>
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Guozhang Wang <gu...@linkedin.com>.
> On July 16, 2014, 11:09 p.m., Guozhang Wang wrote:
> > clients/src/main/java/org/apache/kafka/common/Cluster.java, line 18
> > <https://reviews.apache.org/r/23516/diff/1/?file=632640#file632640line18>
> >
> > Do we ever want to use * in imports?
>
> Jun Rao wrote:
> IDE did the optimization since there are too many imports from the same package.
Yeah. What I am wondering though is if we want to use wildcards in our imports. We used to do so in scala files, which I think messed up some imports, especially with immutable/mutable imports. So I am wondering if we should set it as a coding convention to not use wildcards for non-kafka classes?
- Guozhang
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/#review47940
-----------------------------------------------------------
On July 17, 2014, 4:39 a.m., Jun Rao wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/23516/
> -----------------------------------------------------------
>
> (Updated July 17, 2014, 4:39 a.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1462
> https://issues.apache.org/jira/browse/KAFKA-1462
>
>
> Repository: kafka
>
>
> Description
> -------
>
> address Jay's comments
>
>
> remove partition from all PartitionData since it's redundant
>
>
> minor fixes
>
>
> initial patch
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
> clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
> clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
> clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
> clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
> clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
> clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
> clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
> clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
> clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java PRE-CREATION
> clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
> clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
> core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
> core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
> core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
> core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
> core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
> core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
> core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
> core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
> core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
> core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
> core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
> core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
> core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
> core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
> core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
> core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
> core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
> core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
> core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
>
> Diff: https://reviews.apache.org/r/23516/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Jun Rao
>
>
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Guozhang Wang <gu...@linkedin.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/#review47940
-----------------------------------------------------------
clients/src/main/java/org/apache/kafka/common/Cluster.java
<https://reviews.apache.org/r/23516/#comment84175>
Do we ever want to use * in imports?
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
<https://reviews.apache.org/r/23516/#comment84176>
What is this field used for?
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
<https://reviews.apache.org/r/23516/#comment84177>
I thought the broker will also check consumer id and generation id for offset fetch request?
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
<https://reviews.apache.org/r/23516/#comment84178>
This is a long lasting problem: replica_id here can not only be the id of a follower replica, but also one of the two values: "-1" for ordinary consumer, or "-2" for debugging consumer. We use this field into folds, 1) logging for trouble shooting, which can be helpful only when this is from a follower replica, 2) deciding if it is from the consumer or a replica, for this purpose we do not really care about this id value. We should probably rename this field.
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
<https://reviews.apache.org/r/23516/#comment84179>
Same as above.
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
<https://reviews.apache.org/r/23516/#comment84184>
For the customized rebalancer, will that be supported as: 1) user implement their rebalancer and re-deploy the server with this new code, 2) bounce the consumer setting the new rebalancer? Here I assume the "strategy" will just be the class name?
- Guozhang Wang
On July 15, 2014, 6:36 p.m., Jun Rao wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/23516/
> -----------------------------------------------------------
>
> (Updated July 15, 2014, 6:36 p.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1462
> https://issues.apache.org/jira/browse/KAFKA-1462
>
>
> Repository: kafka
>
>
> Description
> -------
>
> 1. I kept the request objects in the server separated from those in the client. This is because (1) some of the existing request objects are part of old client api (FetechRequest, OffsetCommitRequest, etc) and we can't remove them until the old clients are removed, (2) changing existing request objects on the server side requires significant refactoring.
>
> 2. On the client side, I refactored existing request objects a bit. Now, every request/response object extends from a GenericStruct. GenericStruct provides a standard way for doing serialization and toString so that we don't have to do that on every request. Each request/response can be constructed in two ways: (1) by providing request specific fields; (2) by providing a struct. The latter is used for getting a request/response from its serialized format.
>
> 3. On the server side. What I did is to keep the existing requests more or less untouched. For new types of requests, create a thin wrapper on the server side so that it can leverage the request objects created on the client side. This way the server side object will share the serialization and the toString logic with the client side object. In order to do this, I removed correlationId from the RequestOrResponse object. There is only one place where correlationId is directly referenced and it is not necessary.
>
> 4. Multi-version support. We now need to support two versions of OffsetCommitRequest since for the new consumer work, we added two extra fields in the request. For simplicity, both versions are converted to the same request object. Since the old version doesn't have the new fields, defaults will be used.
>
> 5. The new requests/responses are based on the format described in https://cwiki.apache.org/confluence/display/KAFKA/Kafka+0.9+Consumer+Rewrite+Design#Kafka0.9ConsumerRewriteDesign-Requestformats. I made some minor changes to the wiki so that the new requests follow the current standard.
>
> 6. Added some missing util functions and added unit test for testing the serialization/deserialization logic.
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
> clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
> clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
> clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
> clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
> clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
> clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
> clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
> clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
> clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
> clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
> core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
> core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
> core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
> core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
> core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
> core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
> core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
> core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
> core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
> core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
> core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
> core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
> core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
> core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
> core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
> core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
> core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
> core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
> core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
> core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
> core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
>
> Diff: https://reviews.apache.org/r/23516/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Jun Rao
>
>
Re: Review Request 23516: Patch for KAFKA-1462
Posted by Jun Rao <ju...@gmail.com>.
-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/23516/
-----------------------------------------------------------
(Updated July 15, 2014, 6:36 p.m.)
Review request for kafka.
Bugs: KAFKA-1462
https://issues.apache.org/jira/browse/KAFKA-1462
Repository: kafka
Description (updated)
-------
1. I kept the request objects in the server separated from those in the client. This is because (1) some of the existing request objects are part of old client api (FetechRequest, OffsetCommitRequest, etc) and we can't remove them until the old clients are removed, (2) changing existing request objects on the server side requires significant refactoring.
2. On the client side, I refactored existing request objects a bit. Now, every request/response object extends from a GenericStruct. GenericStruct provides a standard way for doing serialization and toString so that we don't have to do that on every request. Each request/response can be constructed in two ways: (1) by providing request specific fields; (2) by providing a struct. The latter is used for getting a request/response from its serialized format.
3. On the server side. What I did is to keep the existing requests more or less untouched. For new types of requests, create a thin wrapper on the server side so that it can leverage the request objects created on the client side. This way the server side object will share the serialization and the toString logic with the client side object. In order to do this, I removed correlationId from the RequestOrResponse object. There is only one place where correlationId is directly referenced and it is not necessary.
4. Multi-version support. We now need to support two versions of OffsetCommitRequest since for the new consumer work, we added two extra fields in the request. For simplicity, both versions are converted to the same request object. Since the old version doesn't have the new fields, defaults will be used.
5. The new requests/responses are based on the format described in https://cwiki.apache.org/confluence/display/KAFKA/Kafka+0.9+Consumer+Rewrite+Design#Kafka0.9ConsumerRewriteDesign-Requestformats. I made some minor changes to the wiki so that the new requests follow the current standard.
6. Added some missing util functions and added unit test for testing the serialization/deserialization logic.
Diffs
-----
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java a016269512b6d6d6e0fd3fab997e9c8265024eb4
clients/src/main/java/org/apache/kafka/common/Cluster.java c62707ab3aba26771fc4b993df28bf8c44f32309
clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java 6fe7573973832615976defa37fe0dfbb8f911939
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 044b03061802ee5e8ea4f1995fb0988e1a70e9a7
clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java 8cecba50bf067713184208552af36469962cd628
clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ConsumerMetadataResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/GenericStruct.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/HeartbeatRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/HeartbeatResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/JoinGroupRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/JoinGroupResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ListOffsetRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java f35bd87cf0c52a30ed779b25d60bbe64a60b9502
clients/src/main/java/org/apache/kafka/common/requests/MetadataResponse.java 2652c32f123b3bc4b0456d4bc9fbba52c051724c
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java PRE-CREATION
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java 6036f6af1c55c1b0a15471e79b229b17f50ce31c
clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 6cf4fb714916f1a318d788cde8fc0aad9dfb83ca
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java 66cc2fea6443968e525419a203dbc4227e0b1cdf
clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java 257b8287757e40349ea041ed7a651993007a55a8
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 2f98192b064d1ce7c0779e901293edb8c3801915
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java PRE-CREATION
core/src/main/scala/kafka/api/ConsumerMetadataRequest.scala dfad6e6534dd9b00099d110804899080e8d832ab
core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala c72ca14708a3625cb89d5fb92630138d2afa2bf0
core/src/main/scala/kafka/api/ControlledShutdownRequest.scala 7dacb2023788064b736df8b775aaf12281d545b5
core/src/main/scala/kafka/api/ControlledShutdownResponse.scala 46ec3db28f88bbf9e0b0de2133807dc552bcae13
core/src/main/scala/kafka/api/FetchRequest.scala a8b73acd1a813284744359e8434cb52d22063c99
core/src/main/scala/kafka/api/GenericRequestOrResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/HeartbeatRequestAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/HeartbeatResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/JoinGroupRequestAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/JoinGroupResponseAndHeader.scala PRE-CREATION
core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala 3e408174dcc7e8dd9097bae41277ee4f7160afb3
core/src/main/scala/kafka/api/LeaderAndIsrResponse.scala f63644448bb5a1d560f79427284ccbac9d46b789
core/src/main/scala/kafka/api/OffsetCommitRequest.scala 630768ab57afb579049bcbc5d44ee6823b0e7cc2
core/src/main/scala/kafka/api/OffsetCommitResponse.scala 4946e9729ecbf3da35bdab5c832d26977c107e9e
core/src/main/scala/kafka/api/OffsetFetchRequest.scala a32f8588ff02f5fb3c99fb8e5508f462923e8edc
core/src/main/scala/kafka/api/OffsetFetchResponse.scala c1222f422ddb6413bbb2e5da2980903ee70b9156
core/src/main/scala/kafka/api/OffsetRequest.scala 7cbc26c6e38420aa57046a76087fe6d15df72477
core/src/main/scala/kafka/api/OffsetResponse.scala 0e1d6e362a1cec8250cf3930d3046058be4ae192
core/src/main/scala/kafka/api/ProducerRequest.scala 0c295a2fe6712a77cd24719cb42015e2f787b08d
core/src/main/scala/kafka/api/ProducerResponse.scala 5a1d8015379b1f5d9130d9edca89544ee7dd0039
core/src/main/scala/kafka/api/RequestKeys.scala fbfc9d3aeaffed4ca85902125fcc1050086835db
core/src/main/scala/kafka/api/RequestOrResponse.scala 57f87a48c5e87220e7f377b23d2bbfa0d16350dc
core/src/main/scala/kafka/api/StopReplicaRequest.scala 68fc1389ee71122adb716d9d821dd8987a78ecee
core/src/main/scala/kafka/api/StopReplicaResponse.scala c90ddee3d820472236ab554cddd2e0db24233ae3
core/src/main/scala/kafka/api/TopicMetadataRequest.scala a319f2f438bfd84c63edb330685b2b41d4b08aa0
core/src/main/scala/kafka/api/TopicMetadataResponse.scala f6b7429faeab34d0938cb2f78ce91021be7c4b85
core/src/main/scala/kafka/api/UpdateMetadataRequest.scala 543e262b25a946abd84dd58dc5fcee67c6252375
core/src/main/scala/kafka/api/UpdateMetadataResponse.scala c583c1f00c89a993fb9dc280f190c32ea895dca5
core/src/main/scala/kafka/controller/ControllerChannelManager.scala 8763968fbff697e4c5c98ab1274627c192a4d26a
core/src/main/scala/kafka/javaapi/OffsetCommitRequest.scala 08dcc5553ccac7fbec0ea2662b402e2cec079e48
core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala 7e6da164a26b1893c26c624a9998d4fedf8af95e
core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala a2117b34c2ee3554602fe068eed0c90b075958c1
Diff: https://reviews.apache.org/r/23516/diff/
Testing
-------
Thanks,
Jun Rao