kafka判断当前topic内是否包含需要的数据

业务需要判断当前topic内是否包含需要的数据。
当前逻辑是得到该topic所有partitionInfo,startOffset,endOffset.

//get topic partitions info
List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);
List<TopicPartition> topicPartitions = getTopicPartiton(partitionInfoList);

//Get end offset for each partition
Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(topicPartitions);
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions);

然后for循环所有的partition，consumer.seek来去始末的message来判断是否是业务需要。

 for (TopicPartition topicPartition : topicPartitions) {

    // skip when startOffset equal endOffset
    if (beginningOffsets.get(topicPartition).equals(endOffsets.get(topicPartition))) continue;

    //if need seek , must pool fisrt
    consumer.assign(Arrays.asList(topicPartition));
    consumer.poll(duration);

    //Get begin offset message
    consumer.seekToBeginning(Arrays.asList(topicPartition));
    getOffsetValue(consumer);

    //Get end offset message
    consumer.seek(topicPartition, endOffsets.get(topicPartition) - 1);
    getOffsetValue(consumer);
}

Q1:这样玩可以吗？或者有什么好的意见？

Q2:这种情况需要并发吗？还是影响不大，都行