谢谢你的耐心 .
- 在线上向主题添加分区后,kafka使用者停止读取消息,并且不会抛出任何异常 . 消费者只是阻止 . 每次我们都要重启消费者 . 我认为这是不合理的,我找不到任何关于它的文档 .
此外,当处理消息中发生错误时,消费者线程将不会恢复 . 我们的消费者读取消息并将其插入MySql . 一旦网络出现故障,消费者就无法连接到MySql,然后阻止并停止阅读消息,直到我们重新启动它 .
- 添加分区时旧数据和新数据会发生什么变化?文档(https://kafka.apache.org/documentation.html#basic_ops_modify_topic)说:
“请注意,分区的一个用例是在语义上对数据进行分区,并且添加分区不会更改现有数据的分区,因此如果依赖于该分区,这可能会干扰消费者 . 即,如果数据通过散列(密钥)进行分区%number_of_partitions然后这个分区可能会通过添加分区进行洗牌,但Kafka不会尝试以任何方式自动重新分配数据 . “
“不试图自动重新分配数据”是什么意思?旧数据未更改,新数据将不会发送到添加的分区?
- 当经纪人关闭时,kafka 生产环境 者无法发送消息 .
我们有一个主题有3个分区和2个副本 . kafka集群有3个经纪人 . 但是当经纪人倒闭时,会发生异常:
kafka.producer.async.ProducerSendThread.error():103: - Error in handling batch of 65 events
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:90) ~[kafka_2.9.2-0.8.2.0.jar:na]
at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105) [kafka_2.9.2-0.8.2.0.jar:na]
at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88) [kafka_2.9.2-0.8.2.0.jar:na]
at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68) [kafka_2.9.2-0.8.2.0.jar:na]
at scala.collection.immutable.Stream.foreach(Stream.scala:526) [scala-library-2.9.2.jar:na]
at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67) [kafka_2.9.2-0.8.2.0.jar:na]
at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45) [kafka_2.9.2-0.8.2.0.jar:na]
kafka.producer.async.DefaultEventHandler.error():97: - Failed to send requests for topics risk_acts with correlation ids in [433266,433395]
添加新经纪人时也会出现同样的问题 . 我们必须在 生产环境 者的“metadata.broker.list”配置中添加新的代理主机名和端口并重新启动它 .
我们使用的是高级API,而kafka的版本是:
<dependency>
<groupId> org.apache.kafka</groupId >
<artifactId> kafka_2.9.2</artifactId >
<version> 0.8.2.0</version >
</dependency>
生产环境 者配置:
<entry key="metadata.broker.list" value="${metadata.broker.list}" />
<entry key="serializer.class" value="kafka.serializer.StringEncoder" />
<entry key="key.serializer.class" value="kafka.serializer.StringEncoder" />
<entry key="request.required.acks" value="-1" />
<entry key="producer.type" value="async" />
<entry key="queue.enqueue.timeout.ms" value="-1" />
<entry key="compression.codec" value="1" />
消费者配置:
<entry key="zookeeper.connect" value="${zookeeper.connect}" />
<entry key="group.id" value="${kafka.consumer.group.id}" />
<entry key="zookeeper.session.timeout.ms" value="40000" />
<entry key="rebalance.backoff.ms" value="10000" />
<entry key="zookeeper.sync.time.ms" value="2000" />
<entry key="auto.commit.interval.ms" value="1000" />
<entry key="auto.offset.reset" value="smallest" />
生产环境 者代码和消费者代码如下:https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
1 回答
对于#2,请说
key
是Long
. 假设你有10个分区 . 在分区中分发Long
的一种方法是简单地进行模数运算key % num_partitions
. 但现在想想添加分区时会发生什么 . 根据num_partitions
的当前值,已写入的消息将位于错误的分区中 . 这就是说Kafka不会自动为你重新划分任何东西 .