Kafka Broker 领导者变更无效

Question

我有 3 个 kafka 代理，有 3 个分区：

broker.id 1001: 10.18.0.73:9092 领导

broker.id 1002: 10.18.0.73:9093

broker.id 1005：10.18.0.73:9094

Zookeeper 设置为 127.0.0.1:2181

启动：

1001 -> .\kafka-server-start.bat ..\..\config\server.properties
1002 -> .\kafka-server-start.bat ..\..\config\server1.properties
1005 -> .\kafka-server-start.bat ..\..\config\server2.properties

这是 server.properties

broker.id=-1
listeners=PLAINTEXT://10.18.0.73:9092
advertised.listeners=PLAINTEXT://10.18.0.73:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.18.0.73:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
advertised.port=9092
advertised.host.name=10.18.0.73
port=9092

这是 server1.properties

broker.id=-1
listeners=PLAINTEXT://10.18.0.73:9093
advertised.listeners=PLAINTEXT://10.18.0.73:9093
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs4
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
advertised.port=9093
advertised.host.name=10.18.0.73
port=9093

这是 server2.properties

broker.id=-1
listeners=PLAINTEXT://10.18.0.73:9094
advertised.listeners=PLAINTEXT://10.18.0.73:9094
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs2
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
advertised.port=9094
advertised.host.name=10.18.0.73
port=9094

在文件夹 C:\kafka_2.12-2.4.0\config

全部运行

运行生产者

.\kafka-console-producer.bat --broker-list 10.18.0.73:9092,10.18.0.73:9093,10.18.0.73:9094 --topic clinicaleventmanager

运行消费者

.\kafka-console-consumer.bat --bootstrap-server 10.18.0.73:9092,10.18.0.73:9093,10.18.0.73:9094 --topic clinicaleventmanager

我发布一条测试消息

接收成功！

现在，我关闭代理 1001（领导者）

新的领导者是 1002

这条消息在消费者中出现了 1 秒钟，我想这是选举新领导者所需的时间

[2020-01-16 15:33:35,802] WARN [Consumer clientId=consumer-console-consumer-56669-1, groupId=console-consumer-56669] Connection to node 2147482646 (/10.18.0.73:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

如果我尝试发送另一条消息，则该消息不会被消费者读取

新的领导者 1002 似乎没有发送消息。 为什么？

如果我运行 1001 broker.id，一切正常。 谢谢

Answer 1

首先，Kafka 从不“发送（推送）消息”，消费者会要求它们。

其次，您似乎只更改了侦听器、端口和日志目录。

您没有显式创建任何主题，因此您最终会得到一个分区和一个副本的默认值。对于您的主题和内部消费者抵消主题

如果任何副本从您停止的代理中脱机，则无论哪个代理是控制器，其他进程都无法读取（或写入）该副本。

因此，将偏移量（和事务）复制因子更改为 3，然后重试