dajac commented on a change in pull request #10760:
URL: https://github.com/apache/kafka/pull/10760#discussion_r640640161
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -4225,76 +4232,84 @@ public ListOffsetsResult
listOffsets(Map<TopicPartition, OffsetSpec> topicPartit
}
}
- for (final Map.Entry<Node, Map<String, ListOffsetsTopic>> entry :
leaders.entrySet()) {
- final int brokerId = entry.getKey().id();
+ for (final Map.Entry<Node, Map<ListOffsetRequestVersion, Map<String,
ListOffsetsTopic>>> versionedEntry : leaders.entrySet()) {
+ for (final Map.Entry<ListOffsetRequestVersion, Map<String,
ListOffsetsTopic>> entry : versionedEntry.getValue().entrySet()) {
+ final int brokerId = versionedEntry.getKey().id();
- calls.add(new Call("listOffsets on broker " + brokerId,
context.deadline(), new ConstantNodeIdProvider(brokerId)) {
+ calls.add(new Call("listOffsets on broker " + brokerId,
context.deadline(), new ConstantNodeIdProvider(brokerId)) {
- final List<ListOffsetsTopic> partitionsToQuery = new
ArrayList<>(entry.getValue().values());
+ final List<ListOffsetsTopic> partitionsToQuery = new
ArrayList<>(entry.getValue().values());
- @Override
- ListOffsetsRequest.Builder createRequest(int timeoutMs) {
- return ListOffsetsRequest.Builder
+ @Override
+ ListOffsetsRequest.Builder createRequest(int timeoutMs) {
+ ListOffsetRequestVersion requestVersion =
entry.getKey();
+ if (requestVersion ==
ListOffsetRequestVersion.V7AndAbove) {
+ return ListOffsetsRequest.Builder
+
.forMaxTimestamp(context.options().isolationLevel())
+ .setTargetTimes(partitionsToQuery);
+ }
Review comment:
It seems that the current logic send only one request per broker/leader
whereas we could send up to two requests with your PR because specs are
partitioned by `Node` and `ListOffsetRequestVersion`. Previously, they were
only partitioned by `Node`.
Intuitively, I would have approached the problem differently. I would have
put all the specs in the same request and constrained its version to 7 and
above if there is at least one `MAX_TIMESTAMP`. If the request succeeds, all
good. If the request fail with an `UnsupportedVersionException`, I would have
retried with all the specs but the `MAX_TIMESTAMP` ones and I would have failed
the future of the `MAX_TIMESTAMP` specs.
In case of an `UnsupportedVersionException`, the admin client calls the
`handleUnsupportedVersionException` method of the `Call`. This gives you an
opportunity to downgrade and to retry the `Call`. There are couple of example
in the `KafkaAdminClient`.
I wonder if we could rely on a similar pattern and avoid sending two
requests per leader in the worst case. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
