private static void requestAndSetSpecificTimeOffsetsFromKafka ( SimpleConsumer consumer , List < KafkaTopicPartitionState < TopicAndPartition > > partitions , long whichTime ) throws IOException { Map < TopicAndPartition , PartitionOffsetRequestInfo > requestInfo = new HashMap <> ( ) ; for ( KafkaTopicPartitionState < TopicAndPartition > part : partitions ) { requestInfo . put ( part . getKafkaPartitionHandle ( ) , new PartitionOffsetRequestInfo ( whichTime , 1 ) ) ; } requestAndSetOffsetsFromKafka ( consumer , partitions , requestInfo ) ; }
Request offsets before a specific time for a set of partitions via a Kafka consumer .