Вот исключение:
exception:java.nio.channels.ClosedChannelException
все журналы в консоли:
[main] INFO kafka.utils.Log4jControllerRegistration$ - Registered kafka:type=kafka.Log4jController MBean
[main] INFO kafka.utils.VerifiableProperties - Verifying properties
[main] INFO kafka.utils.VerifiableProperties - Property metadata.broker.list is overridden to xxx.xxx.xxx.xxx:6667
[main] INFO kafka.utils.VerifiableProperties - Property request.required.acks is overridden to 1
[main] INFO kafka.utils.VerifiableProperties - Property serializer.class is overridden to kafka.serializer.StringEncoder
[Thread-0] INFO kafka.client.ClientUtils$ - Fetching metadata from broker BrokerEndPoint(0,xxx.xxx.xxx.xxx,6667) with correlation id 0 for 1 topic(s) Set(test)
[Thread-0] INFO kafka.producer.SyncProducer - Connected to xxx.xxx.xxx.xxx:6667 for producing
[Thread-0] INFO kafka.producer.SyncProducer - Disconnecting from xxx.xxx.xxx.xxx:6667
[Thread-0] WARN kafka.client.ClientUtils$ - Fetching topic metadata with correlation id 0 for topics [Set(test)] from broker [BrokerEndPoint(0,xxx.xxx.xxx.xxx,6667)] failed
java.nio.channels.ClosedChannelException
Я нашел несколько ответов в Интернете, которые сказали мне, что я должен установить advertised.host.name в server.properties, но на самом деле я не знаю, какой IP для advertised.host.name установить.
Я полностью потерян в этой ситуации, вот код java, который я пишу в IntelliJ, я просто хотите узнать, какое имя хоста следует указать в BROKER_LIST, чтобы установить соединение между песочницей hortonworks и локальной машиной.
public class KafkaProperties {
public static final String ZK="127.0.0.1:2181";
public static final String TOPIC="test";
public static final String BROKER_LIST= "xx.xxx.xxx.xxx:6667";
}
server.properties:
# Generated by Apache Ambari. Sun Mar 1 19:04:58 2020
auto.create.topics.enable=true
auto.leader.rebalance.enable=true
compression.type=producer
controlled.shutdown.enable=true
controlled.shutdown.max.retries=3
controlled.shutdown.retry.backoff.ms=5000
controller.message.queue.size=10
controller.socket.timeout.ms=30000
default.replication.factor=1
delete.topic.enable=true
external.kafka.metrics.exclude.prefix=kafka.network.RequestMetrics,kafka.server.DelayedOperationPurgatory,kafka.server.BrokerTopicMetrics.BytesRejectedPerSec
external.kafka.metrics.include.prefix=kafka.network.RequestMetrics.ResponseQueueTimeMs.request.OffsetCommit.98percentile,kafka.network.RequestMetrics.ResponseQueueTimeMs.request.Offsets.95percentile,kafka.network.RequestMetrics.ResponseSendTimeMs.request.Fetch.95percentile,kafka.network.RequestMetrics.RequestsPerSec.request
fetch.purgatory.purge.interval.requests=10000
kafka.ganglia.metrics.group=kafka
kafka.ganglia.metrics.host=localhost
kafka.ganglia.metrics.port=8671
kafka.ganglia.metrics.reporter.enabled=true
kafka.metrics.reporters=
kafka.timeline.metrics.host_in_memory_aggregation=
kafka.timeline.metrics.host_in_memory_aggregation_port=
kafka.timeline.metrics.host_in_memory_aggregation_protocol=
kafka.timeline.metrics.hosts=
kafka.timeline.metrics.maxRowCacheSize=10000
kafka.timeline.metrics.port=
kafka.timeline.metrics.protocol=
kafka.timeline.metrics.reporter.enabled=true
kafka.timeline.metrics.reporter.sendInterval=5900
kafka.timeline.metrics.truststore.password=
kafka.timeline.metrics.truststore.path=
kafka.timeline.metrics.truststore.type=
leader.imbalance.check.interval.seconds=300
leader.imbalance.per.broker.percentage=10
listeners=PLAINTEXT://sandbox-hdp.hortonworks.com:6667
log.cleanup.interval.mins=10
log.dirs=/kafka-logs
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760
log.retention.bytes=-1
log.retention.check.interval.ms=600000
log.retention.hours=168
log.roll.hours=168
log.segment.bytes=1073741824
message.max.bytes=1000000
min.insync.replicas=1
num.io.threads=8
num.network.threads=3
num.partitions=1
num.recovery.threads.per.data.dir=1
num.replica.fetchers=1
offset.metadata.max.bytes=4096
offsets.commit.required.acks=-1
offsets.commit.timeout.ms=5000
offsets.load.buffer.size=5242880
offsets.retention.check.interval.ms=600000
offsets.retention.minutes=86400000
offsets.topic.compression.codec=0
offsets.topic.num.partitions=50
offsets.topic.replication.factor=1
offsets.topic.segment.bytes=104857600
port=6667
producer.metrics.enable=false
producer.purgatory.purge.interval.requests=10000
queued.max.requests=500
replica.fetch.max.bytes=1048576
replica.fetch.min.bytes=1
replica.fetch.wait.max.ms=500
replica.high.watermark.checkpoint.interval.ms=5000
replica.lag.max.messages=4000
replica.lag.time.max.ms=10000
replica.socket.receive.buffer.bytes=65536
replica.socket.timeout.ms=30000
sasl.enabled.mechanisms=GSSAPI
sasl.mechanism.inter.broker.protocol=GSSAPI
security.inter.broker.protocol=PLAINTEXT
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
socket.send.buffer.bytes=102400
ssl.client.auth=none
ssl.key.password=
ssl.keystore.location=
ssl.keystore.password=
ssl.truststore.location=
ssl.truststore.password=
zookeeper.connect=sandbox-hdp.hortonworks.com:2181
zookeeper.connection.timeout.ms=25000
zookeeper.session.timeout.ms=30000
zookeeper.sync.time.ms=2000
теперь код в IntelliJ :
public class KafkaProperties {
public static final String ZK="sandbox-hdp.hortonworks.com:2181";
public static final String TOPIC="yanzhao";
public static final String BROKER_LIST= "sandbox-hdp.hortonworks.com:6667";
}
/ ect / hosts file:
127.0.0.1 localhost
::1 localhost ip6-localhost ip6-loopback
fe00::0 ip6-localnet
ff00::0 ip6-mcastprefix
ff02::1 ip6-allnodes
ff02::2 ip6-allrouters
172.18.0.2 sandbox-hdp.hortonworks.com sandbox-hdp
часть jps вывода команды
7846 QuorumPeerMain /usr/hdp/current/zookeeper-server/conf/zoo.cfg
363 AmbariServer
6444 JournalNode
25581 Kafka /usr/hdp/3.0.1.0-187/kafka/config/server.properties
соединение:
[root@sandbox-hdp ~]# netstat -lpn | grep 6667
tcp 0 0 172.18.0.2:6667 0.0.0.0:* LISTEN 25581/java
команда, которую я выполнил в песочнице (настройка потребителя):
kafka-console-consumer.sh --bootstrap-server sandbox-hdp.hortonworks.com:6667 --topic yanzhao
журналы исключений:
[main] INFO kafka.utils.Log4jControllerRegistration$ - Registered kafka:type=kafka.Log4jController MBean
[main] INFO kafka.utils.VerifiableProperties - Verifying properties
[main] INFO kafka.utils.VerifiableProperties - Property metadata.broker.list is overridden to sandbox-hdp.hortonworks.com:6667
[main] INFO kafka.utils.VerifiableProperties - Property request.required.acks is overridden to 1
[main] INFO kafka.utils.VerifiableProperties - Property serializer.class is overridden to kafka.serializer.StringEncoder
[Thread-0] INFO kafka.client.ClientUtils$ - Fetching metadata from broker BrokerEndPoint(0,sandbox-hdp.hortonworks.com,6667) with correlation id 0 for 1 topic(s) Set(yanzhao)
[Thread-0] INFO kafka.producer.SyncProducer - Connected to sandbox-hdp.hortonworks.com:6667 for producing
[Thread-0] INFO kafka.producer.SyncProducer - Disconnecting from sandbox-hdp.hortonworks.com:6667
[Thread-0] WARN kafka.client.ClientUtils$ - Fetching topic metadata with correlation id 0 for topics [Set(yanzhao)] from broker [BrokerEndPoint(0,sandbox-hdp.hortonworks.com,6667)] failed
java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:112)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:63)
at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:83)
at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:76)
at kafka.utils.CoreUtils$.swallow(CoreUtils.scala:85)
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:76)
at kafka.producer.Producer.send(Producer.scala:78)
at kafka.javaapi.producer.Producer.send(Producer.scala:35)
at com.yanzhao.spark.kafka.KafkaProducer.run(KafkaProducer.java:32)
[Thread-0] INFO kafka.producer.SyncProducer - Disconnecting from sandbox-hdp.hortonworks.com:6667
[Thread-0] ERROR kafka.utils.CoreUtils$ - fetching topic metadata for topics [Set(yanzhao)] from broker [ArrayBuffer(BrokerEndPoint(0,sandbox-hdp.hortonworks.com,6667))] failed
kafka.common.KafkaException: fetching topic metadata for topics [Set(yanzhao)] from broker [ArrayBuffer(BrokerEndPoint(0,sandbox-hdp.hortonworks.com,6667))] failed
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:77)
at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:83)
at kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:76)
at kafka.utils.CoreUtils$.swallow(CoreUtils.scala:85)
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:76)
at kafka.producer.Producer.send(Producer.scala:78)
at kafka.javaapi.producer.Producer.send(Producer.scala:35)
at com.yanzhao.spark.kafka.KafkaProducer.run(KafkaProducer.java:32)
Caused by: java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:112)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:80)
at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:63)
... 7 more