碰到了有人问起这个问题,目前java consumer没有利用OffsetAndMetadata中的metadata字段记录提交的时间,故直接通过java consumer来查询是不行,我们需要直接读取__consumer_offsets,代码如下:
public static long lastCommitTimestamp(String groupName, String bootstrapServers) { int partition = Math.abs(groupName.hashCode() % 50); // Assume offset topic partition number is 50 TopicPartition tp = new TopicPartition("__consumer_offsets", partition); Properties props = new Properties(); props.put("bootstrap.servers", bootstrapServers); props.put("group.id", "test"); props.put("enable.auto.commit", "false"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) { consumer.assign(Arrays.asList(tp)); consumer.poll(Duration.ZERO); // take effect Map<TopicPartition, Long> endOffset = consumer.endOffsets(Collections.singletonList(tp)); consumer.seek(tp, endOffset.get(tp) - 1); ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30)); if (records.count() == 0) { throw new IllegalStateException("Consumer group does not commit offsets."); } return records.iterator().next().timestamp(); } }