本文整理了Java中kafka.utils.ZkUtils.getAllTopics()方法的一些代码示例，展示了ZkUtils.getAllTopics()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZkUtils.getAllTopics()方法的具体详情如下：
包路径：kafka.utils.ZkUtils
类名称：ZkUtils
方法名：getAllTopics

ZkUtils.getAllTopics介绍

暂无

代码示例

代码示例来源：origin: confluentinc/kafka-streams-examples

@Override
 public boolean conditionMet() {
  final Set<String> allTopics = new HashSet<>(scala.collection.JavaConversions.seqAsJavaList(zkUtils.getAllTopics()));
  return !allTopics.removeAll(deletedTopics);
 }
}

代码示例来源：origin: org.apache.kafka/kafka_2.10

allTopics.addAll(scala.collection.JavaConversions.seqAsJavaList(zkUtils.getAllTopics()));

代码示例来源：origin: gnuhpc/Kafka-zk-restapi

public List<String> listTopics() {
 return CollectionConvertor.seqConvertJavaList(zkUtils.getAllTopics());
}

代码示例来源：origin: shunfei/DCMonitor

public List<String> getTopics() {
 try {
  return JavaConversions.asJavaList(ZkUtils.getAllTopics(zkClient));
 } catch (Exception e) {
  log.error(e, "could not get topics");
  return Collections.emptyList();
 }
}

代码示例来源：origin: com.cerner.common.kafka/common-kafka-admin

/**
 * Returns the set of all topics in the Kafka cluster
 *
 * @return unmodifiable set of all topics in the Kafka cluster
 *
 * @throws AdminOperationException
 *      if there is an issue retrieving the set of all topics
 */
public Set<String> getTopics() {
  LOG.debug("Retrieving all topics");
  try {
    return Collections.unmodifiableSet(convertToJavaSet(zkUtils.getAllTopics().iterator()));
  } catch (ZkException e) {
    throw new AdminOperationException("Unable to retrieve all topics", e);
  }
}

代码示例来源：origin: SiftScience/kafka-assigner

private static void printCurrentAssignment(ZkUtils zkUtils, List<String> specifiedTopics) {
  Seq<String> topics = specifiedTopics != null ?
      JavaConversions.iterableAsScalaIterable(specifiedTopics).toSeq() :
      zkUtils.getAllTopics();
  System.out.println("CURRENT ASSIGNMENT:");
  System.out.println(
      zkUtils.formatAsReassignmentJson(zkUtils.getReplicaAssignmentForTopics(
          topics)));
}

代码示例来源：origin: apache/atlas

public KafkaBridge(Configuration atlasConf, AtlasClientV2 atlasClientV2) throws Exception {
  String   zookeeperConnect    = getZKConnection(atlasConf);
  int      sessionTimeOutMs    = atlasConf.getInt(ZOOKEEPER_SESSION_TIMEOUT_MS, DEFAULT_ZOOKEEPER_SESSION_TIMEOUT_MS) ;
  int      connectionTimeOutMs = atlasConf.getInt(ZOOKEEPER_CONNECTION_TIMEOUT_MS, DEFAULT_ZOOKEEPER_CONNECTION_TIMEOUT_MS);
  ZkClient zkClient            = new ZkClient(zookeeperConnect, sessionTimeOutMs, connectionTimeOutMs, ZKStringSerializer$.MODULE$);
  this.atlasClientV2   = atlasClientV2;
  this.clusterName     = atlasConf.getString(KAFKA_CLUSTER_NAME, DEFAULT_CLUSTER_NAME);
  this.zkUtils         = new ZkUtils(zkClient, new ZkConnection(zookeeperConnect), JaasUtils.isZkSecurityEnabled());
  this.availableTopics = scala.collection.JavaConversions.seqAsJavaList(zkUtils.getAllTopics());
}

代码示例来源：origin: pinterest/doctorkafka

public static void main(String[] args) throws Exception {
  CommandLine commandLine = parseCommandLine(args);
  String zookeeper = commandLine.getOptionValue(ZOOKEEPER);

  ZkUtils zkUtils = KafkaUtils.getZkUtils(zookeeper);
  Seq<String> topicsSeq = zkUtils.getAllTopics();
  List<String> topics = scala.collection.JavaConverters.seqAsJavaList(topicsSeq);

  scala.collection.mutable.Map<String, scala.collection.Map<Object, Seq<Object>>>
    partitionAssignments = zkUtils.getPartitionAssignmentForTopics(topicsSeq);

  Map<String, Integer> replicationFactors = new HashMap<>();
  Map<String, Integer> partitionCounts = new HashMap<>();

  topics.stream().forEach(topic -> {
   int partitionCount = partitionAssignments.get(topic).get().size();
   int factor = partitionAssignments.get(topic).get().head()._2().size();
   partitionCounts.put(topic, partitionCount);
   replicationFactors.put(topic, factor);
  });

  List<PartitionInfo> urps = KafkaClusterManager.getUnderReplicatedPartitions(
    zookeeper, SecurityProtocol.PLAINTEXT, null, topics, partitionAssignments, replicationFactors, partitionCounts);

  for (PartitionInfo partitionInfo : urps) {
   LOG.info("under-replicated : {}", partitionInfo);
  }
 }
}

代码示例来源：origin: apache/samza

@Override
public List<String> listTables(ExecutionContext context) {
 /**
  * TODO: currently Shell can only talk to Kafka system, but we should use a general way
  *       to connect to different systems.
  */
 lastErrorMsg = "";
 String address = context.getConfigMap().getOrDefault(SAMZA_SQL_SYSTEM_KAFKA_ADDRESS, DEFAULT_SERVER_ADDRESS);
 List<String> tables = null;
 try {
  ZkUtils zkUtils = new ZkUtils(new ZkClient(address, DEFAULT_ZOOKEEPER_CLIENT_TIMEOUT),
    new ZkConnection(address), false);
  tables = JavaConversions.seqAsJavaList(zkUtils.getAllTopics())
   .stream()
   .map(x -> SAMZA_SYSTEM_KAFKA + "." + x)
   .collect(Collectors.toList());
 } catch (ZkTimeoutException ex) {
  String msg = "listTables failed with exception ";
  lastErrorMsg = msg + ex.toString();
  LOG.error(msg, ex);
 }
 return tables;
}

代码示例来源：origin: SiftScience/kafka-assigner

zkUtils.getAllTopics();

代码示例来源：origin: pinterest/doctorkafka

Seq<String> topicsSeq = zkUtils.getAllTopics();
List<String> topics = scala.collection.JavaConverters.seqAsJavaList(topicsSeq);
scala.collection.mutable.Map<String, scala.collection.Map<Object, Seq<Object>>>