KafkaUtilities { public VerifyTopicState verifySupportTopic( KafkaZkClient zkClient, String topic, int expPartitions, int expReplication ) { Objects.requireNonNull(zkClient, "zkClient must not be null"); if (topic == null || topic.isEmpty()) { throw new IllegalArgumentException("topic must not be null or empty"); } if (expPartitions < 1) { throw new IllegalArgumentException("expected partitions must be >= 1"); } if (expReplication < 1) { throw new IllegalArgumentException("expected replication factor must be >= 1"); } VerifyTopicState verifyTopicState = VerifyTopicState.Exactly; try { Set<String> topics = new HashSet<>(); topics.add(topic); scala.Option<scala.collection.immutable.Map<Object, Seq<Object>>> partitionAssignmentOption = zkClient.getPartitionAssignmentForTopics( JavaConversions.asScalaSet(topics).<String>toSet()).get(topic); if (!partitionAssignmentOption.isEmpty()) { scala.collection.Map partitionAssignment = partitionAssignmentOption.get(); int actualNumPartitions = partitionAssignment.size(); if (actualNumPartitions != expPartitions) { log.warn( "The topic {} should have only {} partitions. Having more partitions should " + "not hurt but it is only needed under special circumstances.", topic, expPartitions ); verifyTopicState = VerifyTopicState.Less; } int firstPartitionId = 0; scala.Option<Seq<Object>> replicasOfFirstPartitionOption = partitionAssignment.get(firstPartitionId); if (!replicasOfFirstPartitionOption.isEmpty()) { int actualReplication = replicasOfFirstPartitionOption.get().size(); if (actualReplication < expReplication) { log.warn( "The replication factor of topic {} is {}, which is less than " + "the desired replication factor of {}. If you happen to add more brokers to this" + " cluster, then it is important to increase the replication factor of the " + "topic to eventually {} to ensure reliable and durable metrics collection.", topic, actualReplication, expReplication, expReplication ); verifyTopicState = VerifyTopicState.Less; } } else { log.error("No replicas known for partition 0 of support metrics topic {}", topic); verifyTopicState = VerifyTopicState.Inadequate; } } else { log.error("No partitions are assigned to support metrics topic {}", topic); verifyTopicState = VerifyTopicState.Inadequate; } } catch (Exception e) { log.error("Zookeeper is unavailable. Could not verify topic {}", topic); verifyTopicState = VerifyTopicState.Inadequate; } return verifyTopicState; }  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 verifySupportTopicThrowsIAEWhenKafkaZkClientIsNull() { KafkaUtilities kUtil = new KafkaUtilities(); try { kUtil.verifySupportTopic(null, anyTopic, anyPartitions, anyReplication); fail("IllegalArgumentException expected because zkClient is null"); } catch (NullPointerException e) { } }
@Test public void verifySupportTopicThrowsIAEWhenTopicIsNull() { KafkaUtilities kUtil = new KafkaUtilities(); String nullTopic = null; try { kUtil.verifySupportTopic(mockZkClient, nullTopic, anyPartitions, anyReplication); fail("IllegalArgumentException expected because topic is null"); } catch (IllegalArgumentException e) { } }
@Test public void verifySupportTopicThrowsIAEWhenTopicIsEmpty() { KafkaUtilities kUtil = new KafkaUtilities(); String emptyTopic = ""; try { kUtil.verifySupportTopic(mockZkClient, emptyTopic, anyPartitions, anyReplication); fail("IllegalArgumentException expected because topic is empty"); } catch (IllegalArgumentException e) { } }
@Test public void verifySupportTopicThrowsIAEWhenNumberOfPartitionsIsZero() { KafkaUtilities kUtil = new KafkaUtilities(); int zeroPartitions = 0; try { kUtil.verifySupportTopic(mockZkClient, anyTopic, zeroPartitions, anyReplication); fail("IllegalArgumentException expected because number of partitions is zero"); } catch (IllegalArgumentException e) { } }
@Test public void verifySupportTopicThrowsIAEWhenReplicationFactorIsZero() { KafkaUtilities kUtil = new KafkaUtilities(); int zeroReplication = 0; try { kUtil.verifySupportTopic(mockZkClient, anyTopic, anyPartitions, zeroReplication); fail("IllegalArgumentException expected because replication factor is zero"); } catch (IllegalArgumentException e) { } }