Kafka 流中的更改日志主题 - 设置或更改分区答案

【问题标题】：changelog topic in Kafka streams - setting or altering partitionsKafka 流中的更改日志主题 - 设置或更改分区
【发布时间】：2020-07-10 20:01:59
【问题描述】：

我们有一个流处理器应用程序，它使用具有 n 个分区 (n > 1) 的主题中的数据。

从新开始（没有更新日志主题），开发环境总是创建一个包含 n 个分区的更新日志主题。

在相同的场景中，在生产环境中，分区数始终等于 1，然后我们手动更改为 n 以匹配主题的分区数。

我检查了所有文档，尝试为更改日志设置分区数，但我找不到任何方法。我的最后一个选择是检查更改日志主题是否不存在，然后使用 n 个分区创建它。

由于框架自动创建该主题，是否有任何方法可以设置更改日志的分区数，而无需手动或在代码中创建该主题？

PS：我们使用的是 Kafka 客户端版本 2.3.1。

谢谢，

奥斯汀

【问题讨论】：

标签： apache-kafka stream-processing

【解决方案1】：

我刚刚查看了源代码以了解此功能的详细信息，在撰写本文时，事实证明设置change-logs 主题的分区是不允许的 .

说明

change-logs 主题被归类为内部主题，以下 2 类（InternalTopicConfig 和 InternalTopicManager）有证据表明这一点：

InternalTopicConfig 类的源代码包含以下方法，该方法还表示强制执行此类内部主题的分区数：

public void setNumberOfPartitions(final int numberOfPartitions) {
if (hasEnforcedNumberOfPartitions()) {
    throw new UnsupportedOperationException("number of partitions are enforced on topic " +
                                            "" + name() + " and can't be altered.");
...

InternalTopicManager 类的源代码中的嵌入式文档明确说明了 makeReady() 方法的这一点。

/**
* Prepares a set of given internal topics.
*
* If a topic does not exist creates a new topic.
* If a topic with the correct number of partitions exists ignores it.
* If a topic exists already but has different number of partitions we fail and throw exception requesting user to reset the app before restarting again.
* @return the set of topics which had to be newly created
*/
public Set<String> makeReady(final Map<String, InternalTopicConfig> topics) 
...

正如您在 cmets 中看到的那样，如果这样的主题存在且分区计数正确，它将被忽略，如果分区计数不正确，您将看到错误，建议使用 application reset tool。

希望这会有所帮助！

【讨论】：

真的吗？我们已经成功地手动创建了更改日志主题（ applicationId-stateStoreName-"changelog" ）并分配了正确的分区数来处理我们在生产服务器中遇到的问题。
哦，好的。您能否验证它是否确实以您期望的方式使用它？因为根据代码，应该忽略该主题，否则您应该收到错误 - 除非我正在查看旧版本并且此功能已被修改。
是的，如果我们让应用程序创建更改日志然后分区数 = 1，那么所有消费者都会崩溃。然后我们手动将分区更改为目标编号，然后一切正常。
这很有趣，因为它与源代码中描述的完全不同。

【解决方案2】：

目前我们正在连接到启用 SSL 的 MSK 主题，因此我们没有通过应用程序创建内部主题的写入权限。因此，作为一种解决方法，我们要求 MSK 管理员手动创建具有所需名称的更改日志主题，以便应用程序可以读取它。

此外，目前我们所有的用户主题都是 3 个分区，创建的变更日志主题也有 3 个分区，并具有以下更新的设置。这些设置会派上用场，以防您尝试手动创建更改日志主题（启用压缩以节省空间）：

Configs for Changelog

此外，变更日志主题名称如下所示： (your-application-id)-(materializedAs下的userDefined属性)-changelog

【讨论】：