@Override
    public OffsetMap fetchOffset(KafkaConsumer<?, ?> consumer, String topic) {
        OffsetMap offsetMap = new OffsetMap();

        List<TopicPartition> topicPartitions = new ArrayList<>();

        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);

        partitionInfos.forEach(
                item -> topicPartitions.add(new TopicPartition(topic, item.partition())));

        consumer.assign(topicPartitions);
        consumer.poll(10 * 1000L);

        consumer.seekToEnd(topicPartitions);
        topicPartitions.forEach(
                item ->
                        offsetMap.setLatest(
                                new KafkaTopicPartition(topic, item.partition()),
                                consumer.position(item)));

        consumer.seekToBeginning(topicPartitions);
        topicPartitions.forEach(
                item ->
                        offsetMap.setEarliest(
                                new KafkaTopicPartition(topic, item.partition()),
                                consumer.position(item)));

        return offsetMap;
    }
Reason:
Date: 2018/10/19
Company: www.dtstack.com

@author xuchao
kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer011.java