KafkaUtilities { public List<String> getBootstrapServers(KafkaZkClient zkClient, int maxNumServers) { Objects.requireNonNull(zkClient, "zkClient must not be null"); if (maxNumServers < 1) { throw new IllegalArgumentException("maximum number of requested servers must be >= 1"); } List<Broker> brokers = JavaConversions.seqAsJavaList(zkClient.getAllBrokersInCluster()); if (brokers == null) { return Collections.emptyList(); } else { List<String> bootstrapServers = new ArrayList<>(); for (Broker broker : brokers) { for (EndPoint endPoint : JavaConversions.seqAsJavaList(broker.endPoints())) { if (endPoint.listenerName().value().equals("PLAINTEXT")) { bootstrapServers.add(endPoint.connectionString()); if (bootstrapServers.size() == maxNumServers) { break; } } } } return bootstrapServers; } }  long getNumTopics(KafkaZkClient zkClient); List<String> getBootstrapServers(KafkaZkClient zkClient, int maxNumServers); boolean createAndVerifyTopic(
      KafkaZkClient zkClient,
      String topic,
      int partitions,
      int replication,
      long retentionMs
  ); VerifyTopicState verifySupportTopic(
      KafkaZkClient zkClient,
      String topic,
      int expPartitions,
      int expReplication
  ); boolean isReadyForMetricsCollection(KafkaServer server); boolean isShuttingDown(KafkaServer server);  }
@Test public void getBootstrapServersThrowsIAEWhenKafkaZkClientIsNull() { KafkaUtilities kUtil = new KafkaUtilities(); try { kUtil.getBootstrapServers(null, anyMaxNumServers); fail("IllegalArgumentException expected because zkClient is null"); } catch (NullPointerException e) { } }
@Test public void getBootstrapServersThrowsIAEWhenMaxNumServersIsZero() { KafkaUtilities kUtil = new KafkaUtilities(); int zeroMaxNumServers = 0; try { kUtil.getBootstrapServers(mockZkClient, zeroMaxNumServers); fail("IllegalArgumentException expected because max number of servers is zero"); } catch (IllegalArgumentException e) { } }
@Test public void getBootstrapServersReturnsEmptyListWhenThereAreNoLiveBrokers() { KafkaUtilities kUtil = new KafkaUtilities(); KafkaZkClient zkClient = mock(KafkaZkClient.class); List<Broker> empty = new ArrayList<>(); when(zkClient.getAllBrokersInCluster()).thenReturn(JavaConversions.asScalaBuffer(empty).toList()); assertThat(kUtil.getBootstrapServers(zkClient, anyMaxNumServers)).isEmpty(); }