Kafka
  1. Kafka
  2. KAFKA-349

Create individual "Response" types for each kind of request and wrap them with "BoundedByteBufferSend", remove "xxResponseSend" types for all requests except "FetchRequest"

    Details

    • Type: Improvement Improvement
    • Status: Resolved
    • Priority: Major Major
    • Resolution: Fixed
    • Affects Version/s: None
    • Fix Version/s: 0.8.0
    • Component/s: None
    • Labels:
      None
    1. kafka_responseSend.patch
      33 kB
      Yang Ye
    2. kafka_responseSend.patch.2
      28 kB
      Yang Ye
    3. kafka_responseSend.patch.3
      33 kB
      Yang Ye
    4. kafka_responseSend.patch.4
      32 kB
      Yang Ye
    5. kafka_responseSend.patch.5
      47 kB
      Yang Ye
    6. kafka_responseSend.patch.6
      46 kB
      Yang Ye
    7. kafka_responseSend.patch.7
      69 kB
      Yang Ye
    8. kafka_responseSend.patch.8
      69 kB
      Yang Ye
    9. kafka_responseSend.patch.9
      70 kB
      Yang Ye

      Activity

      Hide
      Yang Ye added a comment - - edited

      1. Replace "Requst" type with "RequestOrResponse" and move it to "kafka.api" from "kafka.network"

      2. Create "FetchResponse", "ProducerResponse", "TopicMetaDataResponse", "OffsetResponse" class

      3. Move the error code inside the response classes

      4. Change the code in "KafkaApis" in response to the change in response formats

      5. Change the Blocking channel receive function to receive just the response (with error code inside) instead of tuple of response and error code.

      6. Remove unnecessary "serialization" and "deserialization" functions

      7. Change corresponding test classes

      Show
      Yang Ye added a comment - - edited 1. Replace "Requst" type with "RequestOrResponse" and move it to "kafka.api" from "kafka.network" 2. Create "FetchResponse", "ProducerResponse", "TopicMetaDataResponse", "OffsetResponse" class 3. Move the error code inside the response classes 4. Change the code in "KafkaApis" in response to the change in response formats 5. Change the Blocking channel receive function to receive just the response (with error code inside) instead of tuple of response and error code. 6. Remove unnecessary "serialization" and "deserialization" functions 7. Change corresponding test classes
      Hide
      Jun Rao added a comment -

      That's a good cleanup patch. Thanks. Some comments:
      1. Maybe we should use this opportunity to make the error code length in all responses consistent (use short int instead of int).
      2. The response classes don't need to store CurrentVersion since their version is always the same as the corresponding request.

      Show
      Jun Rao added a comment - That's a good cleanup patch. Thanks. Some comments: 1. Maybe we should use this opportunity to make the error code length in all responses consistent (use short int instead of int). 2. The response classes don't need to store CurrentVersion since their version is always the same as the corresponding request.
      Hide
      Yang Ye added a comment -

      1. Change all the error code in the system to Short type
      2. Remove the "version id" type from all the response classes

      Show
      Yang Ye added a comment - 1. Change all the error code in the system to Short type 2. Remove the "version id" type from all the response classes
      Hide
      Jun Rao added a comment -

      Thanks for patch v5. One more comment:

      51. FetchResponse,ProducerResponse: We will still need to send the versionId to the client so that the client knows if it gets the right version of the response. We just don't need CurrentVersion in FetchResponse/ProducerResponse since we should just refer to the version constant in FetchRequest/ProducerRequest.

      Show
      Jun Rao added a comment - Thanks for patch v5. One more comment: 51. FetchResponse,ProducerResponse: We will still need to send the versionId to the client so that the client knows if it gets the right version of the response. We just don't need CurrentVersion in FetchResponse/ProducerResponse since we should just refer to the version constant in FetchRequest/ProducerRequest.
      Hide
      Jun Rao added a comment -

      Also, unit tests seem to fail due to the wire protocol changes.

      Show
      Jun Rao added a comment - Also, unit tests seem to fail due to the wire protocol changes.
      Hide
      Yang Ye added a comment -

      1. Added the versionId in ProducerResponse and FetchResponse
      2. Fix one error that leads to test failures. All unit tests been tested through, all passed except two ("testCleanupExpiredSegments" in "LogManagerTest") and "BackwardsCompatibilityTest", but they also fail at the Kafka 0.8 repo. (So it's irrelevant to this patch)

      Show
      Yang Ye added a comment - 1. Added the versionId in ProducerResponse and FetchResponse 2. Fix one error that leads to test failures. All unit tests been tested through, all passed except two ("testCleanupExpiredSegments" in "LogManagerTest") and "BackwardsCompatibilityTest", but they also fail at the Kafka 0.8 repo. (So it's irrelevant to this patch)
      Hide
      Jun Rao added a comment -

      Thanks for patch v6. Some new comments:

      61. FetchResponse, ProducerResponse: VersionId should be the first item is serialized data since we will be depending on VersionId to deserialize the right version of the data in the future.

      62. Could you rebase since trunk has moved?

      63. The only test that fails for me is testCleanupExpiredSegments, which is fixed as part of merging from trunk to 0.8. If you rebase, that test should pass.

      Show
      Jun Rao added a comment - Thanks for patch v6. Some new comments: 61. FetchResponse, ProducerResponse: VersionId should be the first item is serialized data since we will be depending on VersionId to deserialize the right version of the data in the future. 62. Could you rebase since trunk has moved? 63. The only test that fails for me is testCleanupExpiredSegments, which is fixed as part of merging from trunk to 0.8. If you rebase, that test should pass.
      Hide
      Yang Ye added a comment -

      I'm sorry but it's really hard to decouple the RPC patch from the "cleanup" patch, since the latter depends on the first. Jun, can you help check them in all together? This submitted patch include three parts:

      1. Rebase from Neha's latest commit
      2. the cleanup patch
      3. the RPC patch.

      All tests passed.

      Show
      Yang Ye added a comment - I'm sorry but it's really hard to decouple the RPC patch from the "cleanup" patch, since the latter depends on the first. Jun, can you help check them in all together? This submitted patch include three parts: 1. Rebase from Neha's latest commit 2. the cleanup patch 3. the RPC patch. All tests passed.
      Hide
      Jun Rao added a comment -

      Thanks for patch v7. Some comments:

      71. LeaderAndISRRequest:
      71.1 constructor: can we put versionId and clientId before the rest of the fields?
      71.2 writeTo(): we should put versionId first and clientId second.

      72. KafkaApis: in handleLeaderAndISR() and handleStopReplica(), add a TODO comment that the actually logic will be put in later.

      73: ProducerReponse.writeTo: Let's put correlationId before errorcode, to be consistent.

      74. ControllerToBrokerRequestTest seems to fail for me. You need to remove unit. from the package name.

      Show
      Jun Rao added a comment - Thanks for patch v7. Some comments: 71. LeaderAndISRRequest: 71.1 constructor: can we put versionId and clientId before the rest of the fields? 71.2 writeTo(): we should put versionId first and clientId second. 72. KafkaApis: in handleLeaderAndISR() and handleStopReplica(), add a TODO comment that the actually logic will be put in later. 73: ProducerReponse.writeTo: Let's put correlationId before errorcode, to be consistent. 74. ControllerToBrokerRequestTest seems to fail for me. You need to remove unit. from the package name.
      Hide
      Jun Rao added a comment -

      75. ControllerToBrokerRequestTest should use TestUtils.createBrokerConfigs to create broker properties.

      Show
      Jun Rao added a comment - 75. ControllerToBrokerRequestTest should use TestUtils.createBrokerConfigs to create broker properties.
      Hide
      Yang Ye added a comment -

      71. LeaderAndISRRequest:
      71.1 constructor: can we put versionId and clientId before the rest of the fields?

      did that

      71.2 writeTo(): we should put versionId first and clientId second.

      did that

      72. KafkaApis: in handleLeaderAndISR() and handleStopReplica(), add a TODO comment that the actually logic will be put in later.

      did that

      73: ProducerReponse.writeTo: Let's put correlationId before errorcode, to be consistent.

      did that

      74. ControllerToBrokerRequestTest seems to fail for me. You need to remove unit. from the package name.

      did that

      75. ControllerToBrokerRequestTest should use TestUtils.createBrokerConfigs to create broker properties.

      did that

      All tests passed on my local machine

      Show
      Yang Ye added a comment - 71. LeaderAndISRRequest: 71.1 constructor: can we put versionId and clientId before the rest of the fields? did that 71.2 writeTo(): we should put versionId first and clientId second. did that 72. KafkaApis: in handleLeaderAndISR() and handleStopReplica(), add a TODO comment that the actually logic will be put in later. did that 73: ProducerReponse.writeTo: Let's put correlationId before errorcode, to be consistent. did that 74. ControllerToBrokerRequestTest seems to fail for me. You need to remove unit. from the package name. did that 75. ControllerToBrokerRequestTest should use TestUtils.createBrokerConfigs to create broker properties. did that All tests passed on my local machine
      Hide
      Jun Rao added a comment -

      A few more comments for patch v8.

      81. LeaderAndISRRequest and StopReplicaRequest: Could we add a default value for versionId and clientId in the constructor (like in FetchRequest)?

      82. ProducerResponse and OffsetResponse: remove val CurrentVersion.

      Show
      Jun Rao added a comment - A few more comments for patch v8. 81. LeaderAndISRRequest and StopReplicaRequest: Could we add a default value for versionId and clientId in the constructor (like in FetchRequest)? 82. ProducerResponse and OffsetResponse: remove val CurrentVersion.
      Hide
      Yang Ye added a comment -

      81. LeaderAndISRRequest and StopReplicaRequest: Could we add a default value for versionId and clientId in the constructor (like in FetchRequest)?

      Did that

      82. ProducerResponse and OffsetResponse: remove val CurrentVersion.

      Did that

      Show
      Yang Ye added a comment - 81. LeaderAndISRRequest and StopReplicaRequest: Could we add a default value for versionId and clientId in the constructor (like in FetchRequest)? Did that 82. ProducerResponse and OffsetResponse: remove val CurrentVersion. Did that
      Hide
      Jun Rao added a comment -

      Thanks for patch v9. Just committed to 0.8.

      Show
      Jun Rao added a comment - Thanks for patch v9. Just committed to 0.8.

        People

        • Assignee:
          Yang Ye
          Reporter:
          Yang Ye
        • Votes:
          0 Vote for this issue
          Watchers:
          2 Start watching this issue

          Dates

          • Created:
            Updated:
            Resolved:

            Development