カフカのブローカーの接続文字列を取得するには

Question

私はカフカを初めて利用しています。メッセージを送受信するためのコンシューマクラスとプロデューサクラスを実装しようとしました。ブローカのIPとポートのリストである両方のクラスに対して、bootstrap.serversを,で区切って設定する必要があります。例えば、

producerConfig.put("bootstrap.servers", 
       "172.16.20.207:9946,172.16.20.234:9125,172.16.20.36:9636"); 

は、アプリケーションがクラスタのマスターノード上で実行されますので、それだけで Kafka: Get broker host from ZooKeeperへの答えのよう ZooKeeperからブローカーの情報を取得することができるはずです。

public static void main(String[] args) throws Exception { 
    ZooKeeper zk = new ZooKeeper("localhost:2181", 10000, null); 
    List<String> ids = zk.getChildren("/brokers/ids", false); 
    for (String id : ids) { 
     String brokerInfo = new String(zk.getData("/brokers/ids/" + id, false, null)); 
     System.out.println(id + ": " + brokerInfo); 
    } 
} 

しかしこのbrokerInfoは次のようになりJSON形式である： { "jmx_port"： - 1、 "タイムスタンプ"： "1428512949385"、 "ホスト"： "192.168.0.11"、 "バージョン"： 1、 "port"：9093}

この記事では、次の方法で各ブローカの接続文字列を取得し、カンマで結合する方法を提案しました。

for (String id : ids) { 
     String brokerInfoString = new String(zk.getData("/brokers/ids/" + id, false, null)); 
     Broker broker = Broker.createBroker(Integer.valueOf(id), brokerInfoString); 
     if (broker != null) { 
      brokerList.add(broker.connectionString()); 
     } 
    } 

このBrokerクラスがorg.apache.kafka.common.requests.UpdateMetadataRequest.Brokerからのものである場合、それは方法createBrokerとconnectionStringを持っていません。

さらに別の類似のポストが見つかったGetting the list of Brokers Dynamically。しかし、それはhostとportのようなブローカー情報から属性を取得する方法を述べていません。私はおそらく、それらを抽出する文字列のようなjsonのパーサーを書くことができますが、私はそれを行うためにカフカのネイティブな方法があると思う。助言がありますか？

編集：Brokerクラスがkafka.cluster.Brokerであることを認識しました。それでもメソッドconnectionstring()はありません。

Answer 1

あなたは以下のショーとして、クラスタ内のすべてのブローカーの情報を取得するためにZkUtilsを使用することができます。このアプローチに関する一般的な注意事項

ZkUtils zk = ZkUtils.apply("zkHost:2181", 6000, 6000, true); 
List<Broker> brokers = JavaConversions.seqAsJavaList(zk.getAllBrokersInCluster()); 
for (Broker broker : brokers) { 
    //assuming you do not enable security  
    System.out.println(broker.getBrokerEndPoint(SecurityProtocol.PLAINTEXT).host()); 
} 
zk.close();