Я пытаюсь запустить kafka-console-producer.sh. Я набираю что-то на консоль и нажимаю enter, и я получаю тонны трассировок стека на консоль. Оцените любые указатели.
Команда запуска производителя:
./bin/kafka-console-producer.sh --broker-list localhost:2181 --topic test
Затем я набираю строку и нажимаю enter и вижу на консоли следующее.
[2014-11-24 15:27:36,022] ERROR fetching topic metadata for topics [Set(test)] from broker [ArrayBuffer(id:0,host:localhost,port:2181)] failed (kafka.utils.Utils$)
kafka.common.KafkaException: fetching topic metadata for topics [Set(test)] from broker [ArrayBuffer(id:0,host:localhost,port:2181)] failed
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:67)
at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
at kafka.producer.async.DefaultEventHandler$$anonfun$handle$2.apply$mcV$sp(DefaultEventHandler.scala:78)
at kafka.utils.Utils$.swallow(Utils.scala:167)
at kafka.utils.Logging$class.swallowError(Logging.scala:106)
at kafka.utils.Utils$.swallowError(Utils.scala:46)
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104)
at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87)
at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67)
at scala.collection.immutable.Stream.foreach(Stream.scala:547)
at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:66)
at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44)
Caused by: java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
at kafka.utils.Utils$.read(Utils.scala:376)
at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
at kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
at kafka.network.BlockingChannel.receive(BlockingChannel.scala:100)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:74)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:71)
at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
... 12 more
Я также вижу предупреждение на консоли zookeeper:
WARN Exception causing close of session 0x0 due to java.io.EOFException (org.apache.zookeeper.server.NIOServerCnxn)
Я использовал следующее, чтобы настроить тему:
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 1 --replication-factor 1 --topic test
и я могу перечислить свою тему:
./bin/kafka-topics.sh --list --zookeeper localhost:2181
Аргумент broker-list указывает на Zookeeper вместо фактического брокера.
Правильное использование:
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
если ваш брокер работает на порту 9092 (по умолчанию).
Вы можете найти здесь для получения дополнительной информации (ваша проблема описана в Step 4: Send some messages)
Это связано с тем, что вы неправильно используете своего потребителя. Если вы проверите официальную документацию , вы увидите, что вам нужно:
запустить Zookeper
bin/zookeeper-server-start.sh config/zookeeper.properties
запустите Kafka
bin/kafka-server-start.sh config/server.properties
создать тему
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
начать создавать сообщения для темы
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
Обратите внимание на разницу:, которую вы подключаете к zookeper (порт 2181), вам необходимо подключиться к брокеру (порт 9092)