public void seek(long timestamp) {
            // assuming all partitions approximately have the same offset at the given timestamp
            TopicPartition firstPartition = consumers.keySet().iterator().next().firstPartition();
            try {
                ListOffsetsResult.ListOffsetsResultInfo offsetInfo = admin.listOffsets(Map.of(firstPartition, OffsetSpec.forTimestamp(timestamp)))
                    .partitionResult(firstPartition)
                    .get();
                KafkaFuture.allOf(consumers.keySet().stream()
                    .map(topic -> admin.alterConsumerGroupOffsets(groupId(topic), resetOffsetsRequest(topic, offsetInfo.offset())))
                    .map(AlterConsumerGroupOffsetsResult::all)
                    .toArray(KafkaFuture[]::new)).get();
            } catch (ExecutionException | InterruptedException e) {
                throw new RuntimeException("Failed to list or reset consumer offsets", e);
            }
        }
Called when a message is received.

@param topicPartition the topic partition of the received message
@param payload        the received message payload
@param sendTimeNanos  the time in nanoseconds when the message was sent
tools/src/main/java/org/apache/kafka/tools/automq/perf/ConsumerService.java