import org.apache.kafka.clients.consumer.KafkaConsumer; //导入方法依赖的package包/类

/**

* Find the start offsets for the processing windows. We uses kafka 0.10.1.1 that does not support

* KafkaConsumer.

*/

public static Map getProcessingStartOffsets(KafkaConsumer kafkaConsumer,

String brokerStatsTopic,

long startTimestampInMillis) {

List partitionInfos = kafkaConsumer.partitionsFor(brokerStatsTopic);

LOG.info("Get partition info for {} : {} partitions", brokerStatsTopic, partitionInfos.size());

List topicPartitions = partitionInfos.stream()

.map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()))

.collect(Collectors.toList());

Map endOffsets = kafkaConsumer.endOffsets(topicPartitions);

Map beginningOffsets = kafkaConsumer.beginningOffsets(topicPartitions);

Map offsets = new HashMap<>();

for (TopicPartition tp : topicPartitions) {

kafkaConsumer.unsubscribe();

LOG.info("assigning {} to kafkaconsumer", tp);

List tps = new ArrayList<>();

tps.add(tp);

kafkaConsumer.assign(tps);

long endOffset = endOffsets.get(tp);

long beginningOffset = beginningOffsets.get(tp);

long offset = Math.max(endOffsets.get(tp) - 10, beginningOffset);

ConsumerRecord record = retrieveOneMessage(kafkaConsumer, tp, offset);

BrokerStats brokerStats = OperatorUtil.deserializeBrokerStats(record);

if (brokerStats != null) {

long timestamp = brokerStats.getTimestamp();

while (timestamp > startTimestampInMillis) {

offset = Math.max(beginningOffset, offset - 5000);

record = retrieveOneMessage(kafkaConsumer, tp, offset);

brokerStats = OperatorUtil.deserializeBrokerStats(record);

if (brokerStats == null) {

break;

}

timestamp = brokerStats.getTimestamp();

}

}

offsets.put(tp, offset);

LOG.info("{}: offset = {}, endOffset = {}, # of to-be-processed messages = {}",

tp, offset, endOffset, endOffset - offset);

}

return offsets;

}