[GitHub] [kafka] thomaskwscott commented on a change in pull request #10760: KAFKA-12541 Extend ListOffset to fetch offset with max timestamp

[GitBox]

thomaskwscott commented on a change in pull request #10760:
URL: https://github.com/apache/kafka/pull/10760#discussion_r640520669



##########
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:
       At present the only way requestVersion could be V7AndAbove is if we were 
issuing MAX_TIMESTAMP requests because of the way the calls are parsed earlier:
   
   `                    
   ListOffsetRequestVersion requiredRequestVersion = offsetQuery == 
ListOffsetsRequest.MAX_TIMESTAMP
                           ? ListOffsetRequestVersion.V7AndAbove :
                           ListOffsetRequestVersion.V0AndAbove;
   `
   
   All non-max timestamp requests are built using forConsumer rather than 
forMaxTimestamp and so should succeed against older brokers. Maybe the enums 
are a bit misleading in this regard. I'll see if i can come up with something 
better.




-- 
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:
us...@infra.apache.org