[大数据]获取指定时间kafka的offset

    public static void main(String[] args) throws ParseException {

        String topicName = args[0];
        String timeStamp = args[1];//yyyy-MM-dd HH:mm:ss
        String kafkaNode = args[2];//10.1.*.186:9092,10.1.*.187:9092,10.1.*.188:9092

        offsetFind kot = new offsetFind();
        kot.getPartitonAndOffset(topicName,timeStamp,kafkaNode);
    }

    public Map<String, String> getPartitonAndOffset(String topicName,String timeStamp,String kafkaNode) throws ParseException {
        SimpleDateFormat sp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        int port = 9092;//端口


        long timestamp = sp.parse(timeStamp).getTime(); //要查询的时间


        String topic = topicName;//指定主题
        List<String> seeds = new ArrayList<String>(); //kafka broke地址
        seeds.add(kafkaNode.split(",")[0]);
        seeds.add(kafkaNode.split(",")[1]);
        seeds.add(kafkaNode.split(",")[2]);

        offsetFind kot = new offsetFind();
        TreeMap<Integer, PartitionMetadata> metadatas = kot.findLeader(seeds, port, topic);
        Map<String, String> map = new HashMap<String, String>();
        List<Long> offSetList = new ArrayList<>();
        for (Entry<Integer, PartitionMetadata> entry : metadatas.entrySet()) {
            int partition = entry.getKey();
            String leadBroker = entry.getValue().leader().host();
            String clientName = "Client_" + topic + "_" + partition;
            SimpleConsumer consumer = new SimpleConsumer(leadBroker, port, 100000, 64 * 1024, clientName);
            long readOffset = getLastOffset(consumer, topic, partition, timestamp, clientName);
            offSetList.add(readOffset);
            map.put(partition + "", readOffset + "");
            System.out.println(partition + ":" + readOffset);
            if (consumer != null) {
                consumer.close();
            }
        }

        return map;

    }


    public static long getLastOffset(SimpleConsumer consumer, String topic, int partition, long whichTime, String clientName) {
        TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
        Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
        requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
        kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), clientName);
        OffsetResponse response = consumer.getOffsetsBefore(request);

        if (response.hasError()) {
            System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition));
            return 0;
        }
        long[] offsets = response.offsets(topic, partition);
        return offsets[0];//924486286
    }

    private TreeMap<Integer, PartitionMetadata> findLeader(List<String> a_seedBrokers, int a_port, String a_topic) {
        TreeMap<Integer, PartitionMetadata> map = new TreeMap<Integer, PartitionMetadata>();
        loop:
        for (String seed : a_seedBrokers) {
            SimpleConsumer consumer = null;
            try {
                consumer = new SimpleConsumer(seed, a_port, 100000, 64 * 1024, "leaderLookup" + System.currentTimeMillis());
                List<String> topics = Collections.singletonList(a_topic);
                TopicMetadataRequest req = new TopicMetadataRequest(topics);
                kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);

                List<TopicMetadata> metaData = resp.topicsMetadata();
                for (TopicMetadata item : metaData) {
                    for (PartitionMetadata part : item.partitionsMetadata()) {
                        map.put(part.partitionId(), part);
                    }
                }
            } catch (Exception e) {
                System.out.println("Error communicating with Broker [" + seed + "] to find Leader for [" + a_topic + ", ] Reason: " + e);
            } finally {
                if (consumer != null) {
                    consumer.close();
                }
            }
        }
        return map;
    }