Apache kafka 生产集群设置问题

Question

我们一直在尝试在 AWS Linux 机器上建立一个生产级的 Kafka 集群，直到现在我们都没有成功。

卡夫卡版本：2.1.0

机器：

5 r5.xlarge machines for 5 Kafka brokers.
3 t2.medium zookeeper nodes
1 t2.medium node for schema-registry and related tools. (a Single instance of each)
1 m5.xlarge machine for Debezium.

默认代理配置：

num.partitions=15
min.insync.replicas=1
group.max.session.timeout.ms=2000000 
log.cleanup.policy=compact
default.replication.factor=3
zookeeper.session.timeout.ms=30000

我们的问题主要与海量数据有关。我们正在尝试使用 debezium 传输 kafka 主题中的现有表。这些表中的许多都非常庞大，有超过 50000000 行。

到现在为止，我们已经尝试了很多方法，但是我们的集群每次都会因为一个或多个原因而失败。

错误在计划任务“isr-expiration”（kafka.utils.KafkaScheduler）org.apache.zookeeper.KeeperException$SessionExpiredException 中未捕获异常：KeeperErrorCode = Session 在 org.apache 的 /brokers/topics/__consumer_offsets/partitions/0/state 已过期。 zookeeper.KeeperException.create(KeeperException.java:130) 在 org.apache.zookeeper.KeeperException.create(KeeperException.java:54)..

错误 2：

] INFO [Partition xxx.public.driver_operation-14 broker=3] Cached zkVersion [21] 不等于zookeeper，跳过更新ISR (kafka.cluster.Partition) [2018-12-12 14:07:26,551] INFO [分区 xxx.public.hub-14 broker=3] 将 ISR 从 1,3 缩小到 3 (kafka.cluster.Partition) [2018-12-12 14:07:26,556] INFO [分区 xxx.public.hub-14 broker=3] 缓存zkVersion [3] 不等于zookeeper，跳过更新ISR (kafka.cluster.Partition) [2018-12-12 14:07:26,556] INFO [Partition xxx.public.field_data_12_2018-7 broker= 3] 将 ISR 从 1,3 缩小到 3 (kafka.cluster.Partition)

错误 3：

isolationLevel=READ_UNCOMMITTED, toForget=, metadata=(sessionId=888665879, epoch=INITIAL)) (kafka.server.ReplicaFetcherThread) java.io.IOException: 在 org.apache.kafka.clients 读取响应之前，与 3 的连接已断开.NetworkClientUtils.sendAndReceive(NetworkClientUtils.java:97)

还有一些错误：

1. 代理之间频繁断开连接，这可能是在没有自动恢复的情况下不间断地收缩和扩展 ISR 的原因。
2. 架构注册表超时。我不知道架构注册表是如何受到影响的。我没有看到该服务器上的负载过多。我错过了什么吗？我应该将负载均衡器用于架构注册表的多个实例作为故障转移吗？. __schemas 主题中只有 28 条消息。确切的错误消息是 RestClientException：注册操作超时。错误代码：50002

3. 有时消息传输速率超过 100000 条消息/秒，有时下降到 2000 条消息/秒？消息大小会导致这种情况吗？

   为了解决上面的一些问题，我们增加了broker的数量并增加了zookeeper.session.timeout.ms=30000，但我不确定它是否真的解决了我们的问题，如果解决了，如何解决？

我有几个问题：

1. 我们的集群是否足以处理这么多数据。
2. 有什么明显的东西是我们遗漏的吗？
3. 在进入生产级别之前，如何对我的设置进行负载测试？
4. 什么可能导致代理和架构注册表之间的会话超时。
5. 处理架构注册表问题的最佳方法。

我们其中一位经纪人的网络负载。

请随时询问更多信息。

Answer 1

请使用适合您的最新官方版本的Confluence集群。

实际上，您可以通过增加主题的分区数量并增加tasks.max（当然在您的接收器连接器中）来使其更好器中）增加到 1 以上来实现更好的并发性和更快的工作。

请增加Kafka-Connect主题数量并使用Kafka-Connect分布式模式来提高Kafka-connect集群的高可用性。您可以通过在Kafka-Connect和Schema-Registry配置中设置复制因子的数量来实现，例如：

config.storage.replication.factor=2
status.storage.replication.factor=2
offset.storage.replication.factor=2

请将您的大桌子设置topic compression为。snappy它将增加主题的吞吐量，这有助于Debezium连接器更快地工作，并且不使用JSON 转换器，建议使用Avro 转换器！

另外请为您使用负载均衡器另外请为您的架构注册表

为了测试集群，您可以创建一个只有一个表（我的意思是一张大表！）的连接器，并database.whitelist设置snapshot.mode 为initial

以及关于模式注册表！架构注册表用户Kafka和Zookeeper设置这些配置：

bootstrap.servers
kafkastore.connection.url

这就是您的 shema-registry 集群停机的原因