@Override
    public void seek() {
        if (seekTime < 0) {
            return;
        }
        LOGGER.info("start to seek kafka topic {}, seek time is {}", topic, seekTime);
        try {
            Set<TopicPartition> assignedTopicPartitions = consumer.assignment();
            if (assignedTopicPartitions.isEmpty()) {
                LOGGER.error("haven't assigned any topic partitions, do nothing");
                return;
            }
            Map<TopicPartition, Long> timestampsToSearch = assignedTopicPartitions.stream()
                    .collect(Collectors.toMap(tp -> tp, tp -> seekTime));
            Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes(timestampsToSearch);
            List<TopicPartition> endOffsetsTopicPartitions = new ArrayList<>();
            offsetMap.forEach(
                    (tp, offsetAndTimestamp) -> resetOffset(tp, offsetAndTimestamp, endOffsetsTopicPartitions));
            LOGGER.info("topic partition {} should be seek to end", endOffsetsTopicPartitions);
            if (!endOffsetsTopicPartitions.isEmpty()) {
                consumer.seekToEnd(endOffsetsTopicPartitions);
            }
            LOGGER.info("finish to seek kafka topic {}", topic);
        } catch (Throwable t) {
            LOGGER.error("failed to seek kafka topic, ex is {}", t.getMessage(), t);
        }
    }
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