咨询Kafka Streams启动仅获新消息的实现（禁用auto.offset.reset）

针对你的需求（启动时只获取新消息，不依赖auto.offset.reset配置），可以通过预先查询主题最新Offset并手动提交给消费者组的方式实现，具体步骤如下：

核心思路 #

Kafka Streams的消费Offset由应用application.id对应的消费者组管理，默认存储在Kafka内部主题__consumer_offsets中。我们可以在启动Streams应用前，通过AdminClient查询目标主题所有分区的最新Offset（即分区末尾位置），然后将这些Offset手动提交给对应的消费者组，这样应用启动后会直接从最新位置开始消费。

具体代码实现 #

1. 创建AdminClient并查询主题分区信息 #

先初始化AdminClient工具，获取目标主题的所有分区：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.TopicPartition;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

// 初始化AdminClient配置
Properties adminProps = new Properties();
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka集群地址");
AdminClient adminClient = AdminClient.create(adminProps);

// 查询目标主题的分区信息
String targetTopic = "topic";
DescribeTopicsResult topicResult = adminClient.describeTopics(Collections.singletonList(targetTopic));
List<TopicPartition> topicPartitions = new ArrayList<>();
try {
    TopicDescription topicDesc = topicResult.values().get(targetTopic).get();
    topicDesc.partitions().forEach(partitionInfo -> {
        topicPartitions.add(new TopicPartition(targetTopic, partitionInfo.partition()));
    });
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
    adminClient.close();
    return;
}

2. 查询每个分区的最新Offset #

调用AdminClient的listOffsets方法，获取每个分区的末尾Offset：

import org.apache.kafka.clients.admin.OffsetSpec;
import java.util.Map;
import java.util.concurrent.ExecutionException;

// 查询所有分区的最新Offset
Map<TopicPartition, Long> latestOffsets;
try {
    latestOffsets = adminClient.listOffsets(
            topicPartitions.stream()
                    .collect(java.util.stream.Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()))
    ).all().get();
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
    adminClient.close();
    return;
}

3. 将最新Offset提交给消费者组 #

把查询到的Offset转换为OffsetAndMetadata格式，提交给Streams应用对应的消费者组（即application.id的值）：

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

// 替换为你的Streams应用application.id
String consumerGroupId = "你的应用ID";
Map<TopicPartition, OffsetAndMetadata> initialOffsets = new HashMap<>();
latestOffsets.forEach((tp, offset) -> {
    initialOffsets.put(tp, new OffsetAndMetadata(offset));
});

// 提交Offset到消费者组
try {
    adminClient.alterConsumerGroupOffsets(consumerGroupId, initialOffsets).get();
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
} finally {
    adminClient.close();
}

4. 启动Kafka Streams应用 #

完成Offset提交后，正常启动你的Streams应用即可：

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.serialization.Serdes;
import java.util.Properties;

Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, consumerGroupId);
streamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "你的Kafka集群地址");
// 添加序列化/反序列化配置等必要参数
streamsProps.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
streamsProps.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder sb = new StreamsBuilder();
sb.stream(targetTopic).foreach((t, u) -> {
    // 你的消息处理逻辑
});

KafkaStreams streams = new KafkaStreams(sb.build(), streamsProps);
streams.start();

注意事项 #

• 如果应用之前已经运行过并提交过Offset，上述代码会直接覆盖已有Offset，确保启动后从最新位置消费。
• 操作过程中要保证AdminClient拥有alter.group等必要权限。
• 该Offset提交操作仅需在应用启动前执行一次，后续重启时Streams会自动沿用上次提交的Offset。

内容的提问来源于stack exchange，提问作者kimo815