了解 Kafka 分区元数据

Question

我在 NodeJs 应用程序中使用 kafka-node 来通过 loadMetadataForTopics 选项创建主题。我希望我的应用程序能够动态了解可用分区的数量，以便它可以在这些分区之间正确分发消息。

在单节点 Kafka 实例中，该方法创建主题并返回如下元数据：

  "step1_channelOut": {
    "0": {
      "topic": "step1_channelOut",
      "partition": 0,
      "leader": 1,
      "replicas": [
        1
      ],
      "isr": [
        1
      ]
    }
  },

但是在三节点集群中，该方法会创建更多条目：

{
    "0": {
        "topic": "step1_channelOut",
        "partition": 0,
        "leader": 3,
        "replicas": [
            3,
            2,
            1
        ],
        "isr": [
            3,
            2,
            1
        ]
    },
    "1": {
        "topic": "step1_channelOut",
        "partition": 1,
        "leader": 1,
        "replicas": [
            1,
            3,
            2
        ],
        "isr": [
            1,
            3,
            2
        ]
    },
    "2": {
        "topic": "step1_channelOut",
        "partition": 2,
        "leader": 2,
        "replicas": [
            2,
            1,
            3
        ],
        "isr": [
            2,
            1,
            3
        ]
    },
    "3": {
        "topic": "step1_channelOut",
        "partition": 3,
        "leader": 3,
        "replicas": [
            3,
            1,
            2
        ],
        "isr": [
            3,
            1,
            2
        ]
    }
}

在这种情况下，它是否创建了 4 个分区？对我来说它看起来像 - 因为这只是最后一种情况（真正明确地设置分区），只要它是可预测的，我真的不在乎它做了什么。那就是说我控制得越多越好。

zookeeper中的topic信息和kafka server的topic信息是什么关系？有没有更好的方法来通过 nodejs 操作（创建/配置主题）kafka 集群？

为什么是四个分区？我可以理解三个，或者一个，但是四个？

Answer 1

kafka-node 的工作方式是根据您在 server.properties 中找到的全局 Kafka 配置创建主题。检查以下值：

num.partitions=12
default.replication.factor=1

代理数量和分区数量之间没有自动关系。您可以设置 100 个代理但只需要 1 个分区用于主题，或者您可以设置一个具有 1,000 个分区的代理。它们不相关。

没有用于创建主题的非 Java API —— 至少目前还没有。请参阅我之前未回答的问题 here。

如果您想更好地控制主题的创建方式，但仍想使用kafka-node 来实现，您将不得不使用exec 如下命令：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 12 --topic <topic_name_here>

我在node 中这样做：

const exec = require('child_process').exec;

function createTopic(topic, replFactor, numPartitions, cb) {
  var zkHost = "localhost:2181";
  var kafkaHome = "/usr/local/kafka";

  exec(
    `${kafkaHome}/bin/kafka-topics.sh --create --zookeeper ${zkHost} -- replication-factor ${replFactor} --partitions ${numPartitions} --topic ${topic}`,
    (error,stderr,stdout) => cb(topic)
  );
}