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

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

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

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

        Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(topicPartitions);
        beginningOffsets
                .keySet()
                .forEach(
                        item ->
                                offsetMap.setEarliest(
                                        new KafkaTopicPartition(topic, item.partition()),
                                        beginningOffsets.get(item)));

        Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions);
        endOffsets
                .keySet()
                .forEach(
                        item ->
                                offsetMap.setLatest(
                                        new KafkaTopicPartition(topic, item.partition()),
                                        endOffsets.get(item)));

        return offsetMap;
    }
Reason:
Date: 2018/10/19
Company: www.dtstack.com
@author xuchao
kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaConsumer.java