Kafka - 根据主题的前缀设置最小 ISR

Question

我有以下问题要解决。你们知道允许根据主题前缀设置不同的最小 ISR 值的机制吗？例如，使用前缀 app-* 创建的主题将最小 ISR 因子值设置为 2，logs-* 等于 1，但 core-* 的值将是 4。

---

解决方案建议

min.insync.replicas.per.topic=topic_name_1:value,topic_name_2:value 看起来几乎是一种解决方案，但不幸的是，当min.insync.replicas 也出现在kafka.properties 中时，它会被遮蔽。所以我会探索是否有结合这两个道具的解决方案。 source

使用验证最小 ISR 值

./kafka-configs.sh --bootstrap-server localhost:9092 \
 --entity-type topics --entity-name $TOPIC --describe --all

Answer 1

编辑：为了避免 80 年代风格的解决方案（如本答案所示），请看一下 Dirk 的答案。

我只是喜欢 bash/sh

---

对于已经创建的主题，您必须执行partition reassignment operation。

如果您希望控制ISR 用于新创建的主题，您可以将kafka-topics.sh 脚本包装在一个新脚本中，我们称之为createTopics.sh ：

#!/bin/bash
isr=0

if [[ $1 = \logs* ]]; then
        isr=1
elif  [[ $1 = \app* ]]; then
        isr=2
elif  [[ $1 = \core* ]]; then
        isr=4
fi

/bin/kafka-topics.sh --create --topic $1 --replication-factor $isr \
                     --partitions $3  --bootstrap-server $2

• $1 - 主题名称
• $2 - 引导服务器
• $3 - 分区

例如：

  ./createTopics.sh apps-topic localhost:9092 1 && \
  ./createTopics.sh core-topic localhost:9092 1

这将创建两个具有各自复制因子的新主题：

• apps-topic > ISR = 2。
• core-topic > ISR = 4。

（最大 ISR 值，假设所有副本都同步）

---

该脚本还将验证新主题的创建，因为它不会创建任何名称不以下列之一开头的主题：logs / app / core。 Kafka 会显示错误，Replication factor must be larger than 0。

您可以更改此行为以允许使用不同的主题前缀，方法是在脚本开头设置默认的 isr 值而不是 0：例如 isr=1。这将导致：

#!/bin/bash
isr=1

if  [[ $1 = \app* ]]; then
       isr=2
elif  [[ $1 = \core* ]]; then
       isr=4
fi

/bin/kafka-topics.sh --create --topic $1 --replication-factor $isr \
                     --partitions $3  --bootstrap-server $2

Answer 2

您也可以为此使用kafkactl：

# first run with --validate-only to see what kafkactl will do
kafkactl get topics -o compact | grep '^app.*'  | xargs -i echo "echo Topic: {};kafkactl alter topic {} --replication-factor 2 --validate-only" | sh
kafkactl get topics -o compact | grep '^log.*'  | xargs -i echo "echo Topic: {};kafkactl alter topic {} --replication-factor 1 --validate-only" | sh
kafkactl get topics -o compact | grep '^core.*' | xargs -i echo "echo Topic: {};kafkactl alter topic {} --replication-factor 4 --validate-only" | sh

# then do the replica reassignment
kafkactl get topics -o compact | grep '^app.*'  | xargs -i echo "echo Topic: {};kafkactl alter topic {} --replication-factor 2" | sh
kafkactl get topics -o compact | grep '^log.*'  | xargs -i echo "echo Topic: {};kafkactl alter topic {} --replication-factor 1" | sh
kafkactl get topics -o compact | grep '^core.*' | xargs -i echo "echo Topic: {};kafkactl alter topic {} --replication-factor 4" | sh

请注意，kafkactl 用于此目的的 Kafka API 仅适用于 Kafka ≥ 2.4.0。

免责声明：我是这个项目的贡献者