JavaInputDStream<ConsumerRecord<String, String>> stream = KafkaUtils.createDirectStream(streamingContext, "","" ); Set<TopicInfo> topicInfos = new HashSet<>(); stream.foreachRDD(rdd -> { OffsetRange[] offsetRanges = ((HasOffsetRanges) rdd.rdd()).offsetRanges(); topicInfos.addAll(Arrays.stream(offsetRanges).map(offsetRange -> new TopicInfo(offsetRange.topic(),offsetRange.partition(),offsetRange.untilOffset())).collect(Collectors.toSet()));});
1.4m articles
1.4m replys
5 comments
56.9k users