卡夫卡的藏书阁【Book24】- Kafka - KafkaJS Admin 1

“Better to have, and not need, than to need, and not have.”
― Franz Kafka
你不一定要去选择，但你要保有可以做选择的能力、状态

管理员的客户端介面( Admin Client )

管理员的客户端介面可以处理所有丛集相关的动作，像是新增主题(createTopics)、新增分区(createPartitions)...等

const kafka = new Kafka(...)
const admin = kafka.admin()

// 完成动作後记得要中断连线
await admin.connect()
await admin.disconnect()

admin 也适用前几天提到重试机制(retry)

主题清单 ( List topics )

admin 的方法 listTopics 会回传所有主题名称字串组成的阵列，若是遇到问题无法成功回传则会喷错

await admin.listTopics()
// [ 'topic-1', 'topic-2', 'topic-3', ... ]

新增主题 ( Create topics )

admin 的方法 createTopics 回传 true 时表示建立成功、回传 false 表示该主题已经存在，若是遇到问题无法成功处理则会喷错

await admin.createTopics({
    validateOnly: <boolean>,
    waitForLeaders: <boolean>
    timeout: <Number>,
    topics: <ITopicConfig[]>,
})

ITopicConfig 的结构如下：

{
    topic: <String>,
    numPartitions: <Number>,     // 预设值为 1
    replicationFactor: <Number>, // 预设值为 1
    replicaAssignment: <Array>,  // 范例  [{ partition: 0, replicas: [0,1,2] }] - 预设值为 []
    configEntries: <Array>       // 范例 [{ name: 'cleanup.policy', value: 'compact' }] - 预设值为 []
}

• topics
  • 主题的相关设定
  • 无预设值
• validateOnly
  • 如果设定开启，只会验证建立主题是否会成功，但不会真的建立主题
  • 预设值是 false
• timeout
  • 建立主题最长的等待时间 (单位毫秒)
  • 预设值为 5000
• waitForLeaders
  • 如果设定为 true，会等待到主题成功建立，且没有回传错误：LEADER_NOT_AVAILABLE
  • 预设值为 true

删除主题 ( Delete topics )

await admin.deleteTopics({
    topics: <String[]>,
    timeout: <Number>,
})

Apache Kafka 预设是不允许删除主题的，如果需要删除主题的话，要去开启设定、要修改 broker server 的设定档

delete.topic.enable=true