zookeeper重启后Kafka集群丢失消息

Question

我正在使用Docker启动一个kafka代理集群（例如，5个代理，每个容器一个代理）。Kafka版本2.12-0.11.0.0，动物园管理员3.4.10。

场景：

null

tickTime=2000
initLimit=10
syncLimit=5

dataDir=/opt/zookeeper/data

clientPort=2181
maxClientCnxns=10
minSessionTimeout=4000
maxSessionTimeout=1000000
server.1=0.0.0.0:2888:3888

broker.id=1
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://broker1_IP:broker1_PORT
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=127.0.0.1:2181
zookeeper.session.timeout.ms=6000
zookeeper.connection.timeout.ms=1000000
group.initial.rebalance.delay.ms=0

bootstrap.servers=localhost:9092
compression.type=none

zookeeper.connect=127.0.0.1:2181
zookeeper.session.timeout.ms=6000
zookeeper.connection.timeout.ms=1000000
group.id=test-consumer-group

>

在独立模式下启动Zookeeper，然后启动kafka

创建主题

/opt/kafka/bin/kafka-topics.sh--create--zookeeper localhost:2181--replication-factor1--partitions1--topic my-test-topic1

null

• 检查邮件

/opt/kafka/bin/kafka-console-consumer.sh--bootstrap-server localhost:9092--from-bartin--topic my-test-topic1--max-messages1

消息被累犯

null

null

tickTime=2000
initLimit=10
syncLimit=5

dataDir=/opt/zookeeper/data

clientPort=2181
maxClientCnxns=10
minSessionTimeout=4000
maxSessionTimeout=1000000
server.1=0.0.0.0:2888:3888
server.2=broker2_IP:broker2_2888:broker2_3888
server.3=broker3_IP:broker3_2888:broker3_3888
server.4=broker4_IP:broker4_2888:broker4_3888
server.5=broker5_IP:broker5_2888:broker5_3888

server.properties（broker.id唯一，broker_ip:broker_port对于ech broker不同）

broker.id=N
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://broker_IP:broker_PORT
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=127.0.0.1:2181
zookeeper.session.timeout.ms=6000
zookeeper.connection.timeout.ms=1000000
group.initial.rebalance.delay.ms=0

producer.properties在每个代理上从第1到第5

bootstrap.servers=localhost:9092
compression.type=none

Consumer.properties在第1到第5个代理上

zookeeper.connect=127.0.0.1:2181
zookeeper.session.timeout.ms=6000
zookeeper.connection.timeout.ms=1000000
group.id=test-consumer-group

/opt/kafka/bin/kafka-topics.sh--描述--Zookeeper本地主机：2181--主题my-test-topic1主题:my-test-topic1分区计数：1复制因子：1配置：主题:my-test-topic1分区：0领导：5复制：5 ISR：5

是正常行为吗？还是应该留在经纪人1上？

• 检查每个代理上的消息

Answer 1

你试过把售票时间提高到6000点吗？基于Hadoop的设置，他们默认使用这个，声明2000毫秒的设置太低了。我想这同样适用于这里。我现在正在研究一个非常类似Kafka的问题。