Как получить список тем из сервера kafka в Java

Я использую kafka 0.8 версию и очень много нового.

Я хочу узнать список тем, созданных в kafka server, вместе с его метаданные. Есть ли какой-либо API, чтобы узнать это?

В принципе, мне нужно написать Java-пользователя, который должен автоматически обнаружить любую тему в kafka server. Существует API для извлечения TopicMetadata, но для этого требуется имя темы в качестве входного parameters.I нужна информация для всех тем, присутствующих на сервере.

Ответы

Ответ 1

Хорошим местом для начала было бы образцы сценариев, поставляемых с Kafka. В каталоге /bin дистрибутива есть некоторые сценарии оболочки, которые вы можете использовать, один из которых. /kafka -topic-list.sh Если вы запустите это без указания темы, он вернет все темы с их метаданными. Видеть: https://github.com/apache/kafka/blob/0.8/bin/kafka-list-topic.sh

Эта оболочка script в свою очередь запускается: https://github.com/apache/kafka/blob/0.8/core/src/main/scala/kafka/admin/ListTopicCommand.scala

Вышеупомянутые ссылки являются ссылкой на версию версии 0.8 Kafka, поэтому, если вы используете другую версию (даже разницу в точках), обязательно используйте соответствующую ветку/тег для github

Ответ 2

с Кафкой 0,9,0

Вы можете перечислить темы на сервере с помощью предоставленного потребительского метода listTopics();

например.

Map<String, List<PartitionInfo> > topics;

Properties props = new Properties();
props.put("bootstrap.servers", "1.2.3.4:9092");
props.put("group.id", "test-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
topics = consumer.listTopics();
consumer.close();

Ответ 3

Если вы хотите получить информацию о брокере или другой кафке из Zookeeper, тогда kafka.utils.ZkUtils предоставляет хороший интерфейс. Вот код, который я должен перечислить для всех брокеров Zookeeper (там есть множество других методов):

List<Broker> listBrokers() {

        final ZkConnection zkConnection = new ZkConnection(connectionString);
        final int sessionTimeoutMs = 10 * 1000;
        final int connectionTimeoutMs = 20 * 1000;
        final ZkClient zkClient = new ZkClient(connectionString,
                                               sessionTimeoutMs,
                                               connectionTimeoutMs,
                                               ZKStringSerializer$.MODULE$);

        final ZkUtils zkUtils = new ZkUtils(zkClient, zkConnection, false);

        scala.collection.JavaConversions.seqAsJavaList(zkUtils.getAllBrokersInCluster());
}

Ответ 4

Я думаю, что это лучший способ:

ZkClient zkClient = new ZkClient("zkHost:zkPort");
List<String> topics = JavaConversions.asJavaList(ZkUtils.getAllTopics(zkClient));

Ответ 5

Использование Scala:

import java.util.{Properties}
import org.apache.kafka.clients.consumer.KafkaConsumer

object KafkaTest {
  def main(args: Array[String]): Unit = {

    val brokers = args(0)
    val props = new Properties();
    props.put("bootstrap.servers", brokers);
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    val consumer = new KafkaConsumer[String, String](props);
    val topics = consumer.listTopics().keySet();

    println(topics)
  }
}

Ответ 6

Вы можете использовать API Zookeeper для получения списка брокеров, как указано ниже:

    ZooKeeper zk = new ZooKeeper("zookeeperhost, 10000, null);
    List<String> ids = zk.getChildren("/brokers/ids", false);
    List<Map> brokerList = new ArrayList<>();
    ObjectMapper objectMapper = new ObjectMapper();

    for (String id : ids) {
        Map map = objectMapper.readValue(zk.getData("/brokers/ids/" + id, false, null), Map.class);
        brokerList.add(map);
    }

Используйте этот список брокеров, чтобы получить всю тему, используя следующую ссылку

https://cwiki.apache.org/confluence/display/KAFKA/Finding+Topic+and+Partition+Leader