This repository was archived by the owner on Jan 24, 2024. It is now read-only.
Refactor the handler for METADATA request #1107
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
BewareMyPower
changed the title
BewareMyPower
added
release/2.10.0
release/2.8.2.x
release/2.9.2
type/enhancement
eolivelli
reviewed
Feb 24, 2022
Contributor
eolivelli
left a comment
eolivelli
left a comment
There was a problem hiding this comment.
this work seems very interesting.
I did a quick first pass
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
Show resolved
Hide resolved
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
Show resolved
Hide resolved
wenbingshen
suggested changes
Feb 24, 2022
Contributor
wenbingshen
left a comment
wenbingshen
left a comment
There was a problem hiding this comment.
Great work., leave some comments. PTAL.
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
Show resolved
Hide resolved
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
Show resolved
Hide resolved
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
Outdated
Show resolved
Hide resolved
Collaborator
Author
|
Thanks for all your reviews because it's better to have more views on such a refactor, though the existing test already covered much. I've addressed all comments now, PTAL again. @eolivelli @wenbingshen @Demogorgon314 |
wenbingshen
reviewed
Feb 25, 2022
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
Outdated
Show resolved
Hide resolved
Contributor
A great code refactoring! |
wenbingshen
approved these changes
Feb 25, 2022
eolivelli
suggested changes
Feb 25, 2022
Contributor
eolivelli
left a comment
eolivelli
left a comment
There was a problem hiding this comment.
very good
I left one comment about concatList I believe it is better to code it another way
kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/CoreUtils.java
Outdated
Show resolved
Hide resolved
Demogorgon314
approved these changes
Feb 25, 2022
BewareMyPower
added a commit
that referenced
this pull request
Feb 28, 2022
Fixes #1104 ### Motivation In addition to #1104 explained, the existing code has too many nested code. It relies on an atomic integer and multiple callbacks to wait all futures until completed. Since the futures' callbacks are not guaranteed to be executed in a single thread, the helper containers must be thread safe. The better solution is leveraging `CompletableFuture.allOf` method to get all futures' results in a single callback and process the results or handling them to another future. ### Modifications - Add some helper methods into `CoreUtils` to wait all futures and perform conversion among lists and maps. - Add `ListPair` class to avoid null check when processing the result of `groupingBy`. - Add unit tests for methods and class above. - Add `TopicAndMetadata` class as a cache of the full topic name and the number of partitions, which can be negative to indicate an error. - Use a more functional programming code style to rewrite the handler for METADATA request. - Correct an incorrect check for all topics METADATA request, see https://github.com/apache/kafka/blob/8d88b20b2779faa413ffc4c6d2546800e225213f/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java#L162-L164, if `topics` field is empty, only version 0 METADATA request is for all topics. ### TODO The `authorizeTopicsAsync` method and methods from `CoreUtils` can also be used to simplify other request handlers. (cherry picked from commit fbd91a8)
BewareMyPower
added a commit
that referenced
this pull request
Feb 28, 2022
Fixes #1104 ### Motivation In addition to #1104 explained, the existing code has too many nested code. It relies on an atomic integer and multiple callbacks to wait all futures until completed. Since the futures' callbacks are not guaranteed to be executed in a single thread, the helper containers must be thread safe. The better solution is leveraging `CompletableFuture.allOf` method to get all futures' results in a single callback and process the results or handling them to another future. ### Modifications - Add some helper methods into `CoreUtils` to wait all futures and perform conversion among lists and maps. - Add `ListPair` class to avoid null check when processing the result of `groupingBy`. - Add unit tests for methods and class above. - Add `TopicAndMetadata` class as a cache of the full topic name and the number of partitions, which can be negative to indicate an error. - Use a more functional programming code style to rewrite the handler for METADATA request. - Correct an incorrect check for all topics METADATA request, see https://github.com/apache/kafka/blob/8d88b20b2779faa413ffc4c6d2546800e225213f/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java#L162-L164, if `topics` field is empty, only version 0 METADATA request is for all topics. ### TODO The `authorizeTopicsAsync` method and methods from `CoreUtils` can also be used to simplify other request handlers. (cherry picked from commit fbd91a8)
Collaborator
Author
|
This PR will be cherry-picked to branch-2.8.2 after #1120 is merged. |
BewareMyPower
added a commit
that referenced
this pull request
Mar 2, 2022
Fixes #1104 ### Motivation In addition to #1104 explained, the existing code has too many nested code. It relies on an atomic integer and multiple callbacks to wait all futures until completed. Since the futures' callbacks are not guaranteed to be executed in a single thread, the helper containers must be thread safe. The better solution is leveraging `CompletableFuture.allOf` method to get all futures' results in a single callback and process the results or handling them to another future. ### Modifications - Add some helper methods into `CoreUtils` to wait all futures and perform conversion among lists and maps. - Add `ListPair` class to avoid null check when processing the result of `groupingBy`. - Add unit tests for methods and class above. - Add `TopicAndMetadata` class as a cache of the full topic name and the number of partitions, which can be negative to indicate an error. - Use a more functional programming code style to rewrite the handler for METADATA request. - Correct an incorrect check for all topics METADATA request, see https://github.com/apache/kafka/blob/8d88b20b2779faa413ffc4c6d2546800e225213f/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java#L162-L164, if `topics` field is empty, only version 0 METADATA request is for all topics. ### TODO The `authorizeTopicsAsync` method and methods from `CoreUtils` can also be used to simplify other request handlers. (cherry picked from commit fbd91a8)
BewareMyPower
added a commit
that referenced
this pull request
Mar 8, 2022
#1140) ### Motivation After #1107 was cherry-picked into branch-2.8.2, some tests of `KafkaListenerNameTest` are failed. Because that PR forgot to migrate the following code: https://github.com/streamnative/kop/blob/51563feb582424251c8a8345ed9a7606546cf77e/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java#L792-L796 After that when the advertised listener is different with the listener, the brokers field in `MetadataResponse` doesn't contain the advertised listener, while the leader and ISR in `MetadataResponse.PartitionMetadata` use the advertised listener. Then Kafka client won't be able to find the advertised listener in brokers field. ### Modifications Add the partition leader's node to the brokers field if the brokers field doesn't contain the node.
michaeljmarshall
pushed a commit
to michaeljmarshall/kop
that referenced
this pull request
Dec 13, 2022
Fixes streamnative#1104 In addition to streamnative#1104 explained, the existing code has too many nested code. It relies on an atomic integer and multiple callbacks to wait all futures until completed. Since the futures' callbacks are not guaranteed to be executed in a single thread, the helper containers must be thread safe. The better solution is leveraging `CompletableFuture.allOf` method to get all futures' results in a single callback and process the results or handling them to another future. - Add some helper methods into `CoreUtils` to wait all futures and perform conversion among lists and maps. - Add `ListPair` class to avoid null check when processing the result of `groupingBy`. - Add unit tests for methods and class above. - Add `TopicAndMetadata` class as a cache of the full topic name and the number of partitions, which can be negative to indicate an error. - Use a more functional programming code style to rewrite the handler for METADATA request. - Correct an incorrect check for all topics METADATA request, see https://github.com/apache/kafka/blob/8d88b20b2779faa413ffc4c6d2546800e225213f/clients/src/main/java/org/apache/kafka/common/requests/MetadataRequest.java#L162-L164, if `topics` field is empty, only version 0 METADATA request is for all topics. The `authorizeTopicsAsync` method and methods from `CoreUtils` can also be used to simplify other request handlers. (cherry picked from commit fbd91a8) (cherry picked from commit 2d8203f56d3789412dfb60ad93ece585e9e59683)
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Fixes #1104
Motivation
In addition to #1104 explained, the existing code has too many nested code. It relies on an atomic integer and multiple callbacks to wait all futures until completed. Since the futures' callbacks are not guaranteed to be executed in a single thread, the helper containers must be thread safe. The better solution is leveraging
CompletableFuture.allOfmethod to get all futures' results in a single callback and process the results or handling them to another future.Modifications
CoreUtilsto wait all futures and perform conversion among lists and maps.ListPairclass to avoid null check when processing the result ofgroupingBy.TopicAndMetadataclass as a cache of the full topic name and the number of partitions, which can be negative to indicate an error.topicsfield is empty, only version 0 METADATA request is for all topics.TODO
The
authorizeTopicsAsyncmethod and methods fromCoreUtilscan also be used to simplify other request handlers.