private Map<TopicPartition, OffsetAndMetadata> resetOffsetsRequest(Topic topic, long offset) {
            return topic.partitions().stream()
                .collect(Collectors.toMap(
                    partition -> partition,
                    ignore -> new OffsetAndMetadata(offset)
                ));
        }
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