private void resetOffset(
            TopicPartition tp,
            OffsetAndTimestamp offsetAndTimestamp,
            List<TopicPartition> endOffsetsTopicPartitions) {
        // if offsetAndTimestamp = null, means the time you seek is later than the last offset
        if (offsetAndTimestamp == null) {
            LOGGER.info("tp {} has null offsetAndTimestamp, reset to end", tp);
            endOffsetsTopicPartitions.add(tp);
            return;
        }
        long expected = offsetAndTimestamp.offset();
        long last = consumer.position(tp);
        LOGGER.info("for tp {}, expected offset is {}, last offset is {}", tp, expected, last);
        // only reset if last consume offset earlier than the expected one
        if (last < expected) {
            LOGGER.info("do seek for tp {}", tp);
            consumer.seek(tp, offsetAndTimestamp.offset());
            long afterSeek = consumer.position(tp);
            LOGGER.info("after seek, the offset for tp {} is {}", tp, afterSeek);
        }

    }
Kafka seeker.
Only seek those topics and partitions assigned by server.
The whole process can be divided into three steps:
1. Find all topics and partitions of this consumer;
2. Calculate the offsets of the excepted seek time of each topic and partition respectively;
3. Reset these topics and partitions to the corresponding offset
inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/fetcher/kafka/KafkaSeeker.java