MINOR: Fix an uncompatible bug in GetOffsetShell (apache#11936)

In KIP-815 we replaced KafkaConsumer with AdminClient in GetOffsetShell. In the previous implementation, partitions were just ignored if there is no offset for them, however, we will print -1 instead now, This PR fix this inconsistency.

Reviewers: David Jacot <[email protected]>, Luke Chen <[email protected]>

core/src/main/scala/kafka/tools/GetOffsetShell.scala

@@ -22,7 +22,7 @@ import joptsimple._
 import kafka.utils.{CommandLineUtils, Exit, IncludeList, ToolsUtils}
 import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, ListTopicsOptions, OffsetSpec}
 import org.apache.kafka.common.{KafkaException, TopicPartition}
-import org.apache.kafka.common.requests.ListOffsetsRequest
+import org.apache.kafka.common.requests.{ListOffsetsRequest, ListOffsetsResponse}
 import org.apache.kafka.common.utils.Utils
 
 import java.util.Properties
@@ -135,7 +135,11 @@ object GetOffsetShell {
       val partitionOffsets = partitionInfos.flatMap { tp =>
         try {
           val partitionInfo = listOffsetsResult.partitionResult(tp).get
-          Some((tp, partitionInfo.offset))
+          if (partitionInfo.offset != ListOffsetsResponse.UNKNOWN_OFFSET) {
+            Some((tp, partitionInfo.offset))
+          } else {
+            None
+          }
         } catch {
           case e: ExecutionException =>
             e.getCause match {

core/src/test/scala/kafka/tools/GetOffsetShellTest.scala

@@ -166,6 +166,13 @@ class GetOffsetShellTest extends KafkaServerTestHarness with Logging {
     )
   }
 
+  @Test
+  def testNoOffsetIfTimestampGreaterThanLatestRecord(): Unit = {
+    val time = (System.currentTimeMillis() * 2).toString
+    val offsets = executeAndParse(Array("--topic-partitions", "topic.*", "--time", time))
+    assertEquals(List.empty, offsets)
+  }
+
   @Test
   def testTopicPartitionsArgWithInternalExcluded(): Unit = {
     val offsets = executeAndParse(Array("--topic-partitions",