“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