dengziming commented on a change in pull request #11173:
URL: https://github.com/apache/kafka/pull/11173#discussion_r763688712
##########
File path: core/src/main/scala/kafka/tools/GetOffsetShell.scala
##########
@@ -140,22 +139,25 @@ object GetOffsetShell {
Some(new TopicPartition(p.topic, p.partition))
}
+ val offsetSpec = listOffsetsTimestamp match {
+ case ListOffsetsRequest.EARLIEST_TIMESTAMP => OffsetSpec.earliest()
+ case ListOffsetsRequest.LATEST_TIMESTAMP => OffsetSpec.latest()
+ case ListOffsetsRequest.MAX_TIMESTAMP => OffsetSpec.maxTimestamp()
+ case _ => OffsetSpec.forTimestamp(listOffsetsTimestamp)
+ }
+
+ val timestampsToSearch = topicPartitions.map(tp => tp ->
offsetSpec).toMap.asJava
+
/* Note that the value of the map can be null */
- val partitionOffsets: collection.Map[TopicPartition, java.lang.Long] =
listOffsetsTimestamp match {
- case ListOffsetsRequest.EARLIEST_TIMESTAMP =>
consumer.beginningOffsets(topicPartitions.asJava).asScala
- case ListOffsetsRequest.LATEST_TIMESTAMP =>
consumer.endOffsets(topicPartitions.asJava).asScala
- case _ =>
- val timestampsToSearch = topicPartitions.map(tp => tp ->
(listOffsetsTimestamp: java.lang.Long)).toMap.asJava
- consumer.offsetsForTimes(timestampsToSearch).asScala.map { case (k,
x) =>
- if (x == null) (k, null) else (k, x.offset: java.lang.Long)
- }
+ val partitionOffsets =
client.listOffsets(timestampsToSearch).all().get().asScala.map { case (k, x) =>
+ if (x == null) (k, null) else (k, x.offset: java.lang.Long)
Review comment:
Yes, this is a good idea, after moving the leader check to
`ListOffsetsResult.partitionResult`, we can use `TopicPartition` in
`TopicFilter`
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
