KafkaUtilities { public boolean createAndVerifyTopic( KafkaZkClient zkClient, String topic, int partitions, int replication, long retentionMs ) { Objects.requireNonNull(zkClient, "zkClient must not be null"); validateTopicParams(topic, partitions, replication, retentionMs); boolean topicCreated = true; try { if (zkClient.topicExists(topic)) { return ( verifySupportTopic(zkClient, topic, partitions, replication) != VerifyTopicState.Inadequate ); } Seq<Broker> brokerList = zkClient.getAllBrokersInCluster(); int actualReplication = Math.min(replication, brokerList.size()); if (actualReplication < replication) { log.warn( "The replication factor of topic {} will be set to {}, which is less than the " + "desired replication factor of {} (reason: this cluster contains only {} brokers). " + "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, replication, brokerList.size(), replication ); } Properties metricsTopicProps = new Properties(); metricsTopicProps.put(LogConfig.RetentionMsProp(), String.valueOf(retentionMs)); log.info("Attempting to create topic {} with {} replicas, assuming {} total brokers", topic, actualReplication, brokerList.size() ); AdminZkClient adminClient = new AdminZkClient(zkClient); adminClient.createTopic( topic, partitions, actualReplication, metricsTopicProps, Disabled$.MODULE$ ); for (int part = 0; part < partitions; ++part) { waitUntilLeaderIsElected(zkClient, topic, part, 30000L); } } catch (TopicExistsException te) { log.info("Topic {} already exists", topic); topicCreated = false; } catch (AdminOperationException e) { topicCreated = false; log.error("Could not create topic {}: {}", topic, e.getMessage()); } catch (TimeoutException toe) { topicCreated = false; log.error("Timed out waiting for leader to be elected after creating topic: {}", toe.getMessage()); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); topicCreated = false; log.error("Interrupted the wait for leader to be elected after creating topic={}", topic); } catch (Exception e) { topicCreated = false; log.error("Zookeeper is unavailable. Could not create topic {}: {}", topic, e.getMessage()); } return topicCreated; }  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 createTopicThrowsIAEWhenTopicIsEmpty() { KafkaUtilities kUtil = new KafkaUtilities(); String emptyTopic = ""; try { kUtil.createAndVerifyTopic(mockZkClient, emptyTopic, anyPartitions, anyReplication, anyRetentionMs); fail("IllegalArgumentException expected because topic is empty"); } catch (IllegalArgumentException e) { } }
@Test public void createTopicThrowsIAEWhenNumberOfPartitionsIsZero() { KafkaUtilities kUtil = new KafkaUtilities(); int zeroPartitions = 0; try { kUtil.createAndVerifyTopic(mockZkClient, anyTopic, zeroPartitions, anyReplication, anyRetentionMs); fail("IllegalArgumentException expected because number of partitions is zero"); } catch (IllegalArgumentException e) { } }
@Test public void createTopicThrowsIAEWhenReplicationFactorIsZero() { KafkaUtilities kUtil = new KafkaUtilities(); int zeroReplication = 0; try { kUtil.createAndVerifyTopic(mockZkClient, anyTopic, anyPartitions, zeroReplication, anyRetentionMs); fail("IllegalArgumentException expected because replication factor is zero"); } catch (IllegalArgumentException e) { } }
@Test public void createTopicThrowsIAEWhenRetentionMsIsZero() { KafkaUtilities kUtil = new KafkaUtilities(); long zeroRetentionMs = 0; try { kUtil.createAndVerifyTopic(mockZkClient, anyTopic, anyPartitions, anyReplication, zeroRetentionMs); fail("IllegalArgumentException expected because retention.ms is zero"); } catch (IllegalArgumentException e) { } }
@Test public void createTopicFailsWhenThereAreNoLiveBrokers() { KafkaUtilities kUtil = new KafkaUtilities(); EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(); cluster.startCluster(1); KafkaServer broker = cluster.getBroker(0); KafkaZkClient defunctZkClient = broker.zkClient(); cluster.stopCluster(); assertThat(kUtil.createAndVerifyTopic(defunctZkClient, anyTopic, anyPartitions, anyReplication, anyRetentionMs)).isFalse(); }
@Test public void leaderIsElectedAfterCreateTopicReturns() { KafkaUtilities kUtil = new KafkaUtilities(); EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(); int numBrokers = 3; cluster.startCluster(numBrokers); KafkaServer broker = cluster.getBroker(0); KafkaZkClient zkClient = broker.zkClient(); int replication = numBrokers; assertThat(kUtil.createAndVerifyTopic(zkClient, anyTopic, anyPartitions, replication, oneYearRetention)).isTrue(); assertThat(zkClient.getLeaderForPartition( new TopicPartition(anyTopic, 0)).isDefined()).isTrue(); }
@Test public void createTopicThrowsIAEWhenKafkaZkClientIsNull() { KafkaUtilities kUtil = new KafkaUtilities(); try { kUtil.createAndVerifyTopic(null, anyTopic, anyPartitions, anyReplication, anyRetentionMs); fail("IllegalArgumentException expected because zkClient is null"); } catch (NullPointerException e) { } }
@Test public void createTopicThrowsIAEWhenTopicIsNull() { KafkaUtilities kUtil = new KafkaUtilities(); String nullTopic = null; try { kUtil.createAndVerifyTopic(mockZkClient, nullTopic, anyPartitions, anyReplication, anyRetentionMs); fail("IllegalArgumentException expected because topic is null"); } catch (IllegalArgumentException e) { } }