Scala编写的Kafka Consumer无输出问题求助

先把你提供的代码片段整理出来（看起来代码没写完，有截断）：

val topicProducer = "testOutput" 
val props = new Properties() 
props.put("bootstrap.servers","host:9092,host:9092") 
props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer") 
props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer") 
props.put("group.id", "test"); 
val kafkaConsumer = new KafkaConsumer[String, String](props... // 这里代码被截断了

控制台空白通常是因为核心消费逻辑缺失或者配置/环境问题，我给你列几个排查方向和修复方案：

• 核心消费逻辑未实现
  你目前只初始化了KafkaConsumer，但没有订阅主题，也没有启动拉取消息的循环。这是最常见的原因！完整的消费者逻辑需要包含：订阅主题、循环调用poll()拉取消息、处理消息这几个步骤。

• 配置项拼写错误
  最后一行代码里你写的是new KafkaConsumer[String, String](props...，看起来变量名可能写错了？应该是props而不是prop，如果变量名错误会导致初始化失败，自然没有输出。

• 主题无消息或消费位移问题

  • 先确认testOutput主题里有没有消息，可以用Kafka命令行工具验证：kafka-console-consumer.sh --bootstrap-server host:9092 --topic testOutput --from-beginning
  • 如果你的消费者组test之前已经消费过该主题的所有消息，并且位移已经提交，默认情况下新的消费者会从最新位移开始拉取，这时如果主题没有新消息，控制台就会空白。可以添加配置props.put("auto.offset.reset", "earliest")，让消费者从最早的位移开始拉取历史消息。

• 网络或权限问题

  • 检查bootstrap.servers配置的地址是否能正常连通，比如用telnet host 9092测试端口是否开放
  • 确认你的消费者有访问testOutput主题的权限，比如是否配置了正确的ACL规则

• 序列化/反序列化不匹配
  确保生产者使用的是StringSerializer，如果生产者用了其他序列化器（比如JsonSerializer），而消费者用StringDeserializer会导致反序列化失败，消息无法正常输出。

完整的Scala Kafka Consumer示例代码 #

import org.apache.kafka.clients.consumer.{KafkaConsumer, ConsumerRecords, ConsumerRecord}
import java.util.Properties
import java.time.Duration

object KafkaConsumerDemo {
  def main(args: Array[String]): Unit = {
    val topic = "testOutput"
    val props = new Properties()
    props.put("bootstrap.servers", "host:9092,host:9092")
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("group.id", "test")
    // 从最早位移开始拉取（可选，根据需求调整）
    props.put("auto.offset.reset", "earliest")

    val consumer = new KafkaConsumer[String, String](props)
    // 订阅目标主题
    consumer.subscribe(java.util.Collections.singletonList(topic))

    try {
      while (true) {
        // 拉取消息，超时时间设置为1秒
        val records: ConsumerRecords[String, String] = consumer.poll(Duration.ofMillis(1000))
        // 遍历处理并打印消息
        import scala.collection.JavaConverters._
        for (record <- records.asScala) {
          println(s"Received message: key = ${record.key()}, value = ${record.value()}, partition = ${record.partition()}, offset = ${record.offset()}")
        }
      }
    } finally {
      // 程序退出前关闭消费者
      consumer.close()
    }
  }
}

你可以按照上面的步骤逐一排查，先补全核心消费逻辑，再检查配置和环境问题，应该就能解决控制台空白的问题了。

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