KafkaStreams无法从Avro序列化主题流式传输数据求助

核心问题分析 #

你的Kafka Streams应用启动后无输出、未订阅目标主题，而普通消费者可以正常工作，主要原因有两个：

1. 未配置自动偏移重置策略，新消费组默认从latest偏移量开始消费，若主题没有新产生的消息，自然不会有输出。
2. 缺少应用生命周期管理，Kafka Streams的start()是异步执行，main方法执行完成后可能导致JVM意外退出（虽然你提到应用没崩溃，但添加Shutdown Hook是标准操作）。

修复后的代码 #

public static void main(String[] args) {
    // 定义流配置
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "Test-stream-UserRegistrationServicebbb");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
    // 关键：添加自动偏移重置配置，让新消费组从头消费已有消息
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    // 利用全局配置初始化Serde，无需单独重复配置schema registry地址
    SpecificAvroSerde<Pending_Registrations> pendingRegistrationsSerde = new SpecificAvroSerde<>();
    // 第二个参数设为false表示这是value serde（key serde设为true）
    pendingRegistrationsSerde.configure(props, false);

    StreamsBuilder builder = new StreamsBuilder();
    // 创建流并指定serde
    KStream<String, Pending_Registrations> userStream = builder.stream("User.Pending-Registrations",
            Consumed.with(Serdes.String(), pendingRegistrationsSerde));
    // 打印流数据
    userStream.foreach((key, value) -> System.out.println("key: " + key + " value: " + value));

    KafkaStreams streams = new KafkaStreams(builder.build(), props);
    
    // 添加Shutdown Hook，确保优雅关闭
    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    
    streams.start();
}

关键修改说明 #

1. 添加自动偏移重置配置：
   普通消费者显式设置了AUTO_OFFSET_RESET_CONFIG为earliest，但Kafka Streams默认不会自动配置。对于新的application.id（即新消费组），必须手动指定该参数才能消费主题中已有的历史消息。
2. 简化Serde配置：
   已经在全局Properties中配置了SCHEMA_REGISTRY_URL_CONFIG，无需再单独给Serde创建serdeConfig映射，直接传入全局配置即可，避免冗余配置。
3. 添加Shutdown Hook：
   确保应用在收到终止信号时能优雅关闭Kafka Streams，同时防止main方法执行完毕后JVM立即退出（虽然你的应用没崩溃，但这是生产环境的标准做法）。

额外验证步骤 #

• 用Kafka命令行工具确认主题存在且有消息：

  # 查看主题列表
  kafka-topics.sh --list --bootstrap-server localhost:9092
  # 查看主题消息数量
  kafka-run-class.sh kafka.tools.GetOffsetShell --topic User.Pending-Registrations --time -1 --bootstrap-server localhost:9092
  

• 检查Control Center中消费组Test-stream-UserRegistrationServicebbb的状态，确认是否已订阅目标主题。

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