Kafka制作人创建主题但无法发送消息

Question

我是斯卡拉和Kafka的新手，遇到了一些麻烦。

我正在尝试将scala kafka producer连接到安装在cloudera express服务器上的kafka服务器。我已经用这些指令在VMs中这样做过一次了，没有任何问题。

当我运行producer时，所需的主题被创建，但没有任何消息被发送，或者我是这样认为的。

Kafka制作人

import java.util.Properties
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

class KafkaProducerManager {

    val props = new Properties()
    props.put("bootstrap.servers", KafkaServer.KAFKA_ADDRESS)
    props.put("acks", "all")
    props.put("retries", "2")
    props.put("auto.commit.interval.ms", "1000")
    props.put("linger.ms", "1")
    props.put("block.on.buffer.full", "true")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("auto.create.topics.enable", "true")

    val producer = new KafkaProducer[String, String](props)

   def startCounter() {
       println("Start Producer Counter")
       for (i <- 1 to 100) {
           producer.send(new ProducerRecord("test-counter", i.toString, "Package " + i))
           println("Producer - Send: " + i)
       }

       println("Closing producer")
       producer.close()
   }
}

当我执行run方法时，我看到“producer-send:#”作为该方法的输出，并且没有错误。所以我假设这段代码已经把信息发送给了Kafka。

在运行Producer之前，我在kafka服务器上启动了以下操作：

 kafka-console-consumer --zookeeper zk:2181 --topic test-counter

kafka-topics -zookeeper zk:2181 --list

import java.util.{Arrays, Properties}
import org.apache.kafka.clients.consumer.KafkaConsumer

class KafkaConsumerManager {

    val props = new Properties()
    props.put("bootstrap.servers", KafkaServer.KAFKA_ADDRESS)
    props.put("group.id", "testGroup")
    props.put("enable.auto.commit", "true")
    props.put("auto.commit.interval.ms", "1000")
    props.put("linger.ms", "1")
    props.put("session.timeout.ms", "3000")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("zookeeper.connect", KafkaServer.ZOOKEEPER_ADDRESS)

    val consumer = new KafkaConsumer[String, String](props)

    def start() {
        println("Start Consumer")
        consumer.subscribe(Arrays.asList("test-counter"))

        while (true) {
            val records = consumer.poll(100)
            val iterator = records.iterator()

            while (iterator.hasNext) {
                val record = iterator.next()
                printf("Consumer: offset = %d, key = %s, value = %s \n", record.offset(), record.key(), record.value())
            }
        }
    }
}

当我通过kafka-console-producer在服务器上创建消息时，我看到它们出现在服务器上的kafka-console-consumer中，但不出现在我编写的consumer中。

kafka-console-producer --broker-list ks:9092 --topic test-counter

kafkaserver.zookeeper_address与具有kafka-console-consumer的参数zk:2181相同，kafkaserver.kafka_address与具有kafka-console-producer的参数ks:9092相同。

Answer 1

我试了一下代码，发现：

>

应在使用者属性中指定键和值反序列化器：

 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")

session.timeout.ms属性有问题。从这里：