KafkaOperations implements AutoCloseable { public boolean createTopic(final CreateTopic createTopic) { final NewTopic newTopic = new NewTopic( createTopic.getName(), createTopic.getNumberOfPartitions(), createTopic.getReplicaFactor() ); try { final CreateTopicsResult result = adminClient.createTopics(Collections.singleton(newTopic)); result.all().get(); return true; } catch (final ExecutionException e) { throw handleExecutionException(e); } catch (final InterruptedException exception) { throw new RuntimeException(exception.getMessage(), exception); } } KafkaOperations(final AdminClient adminClient, final KafkaConsumer<String, String> consumerClient); TopicList getAvailableTopics(); NodeList getClusterNodes(); Map<String, TopicDetails> getTopicDetails(final Collection<String> topics); TopicDetails getTopicDetails(final String topic); TopicConfig getTopicConfig(final String topic); BrokerConfig getBrokerConfig(final String brokerId); boolean createTopic(final CreateTopic createTopic); TopicConfig alterTopicConfig(final String topic, final Map<String, String> configItems); boolean removeTopic(final String topic); List<ConsumerGroupIdentifier> listConsumers(); boolean removeConsumerGroup(final String id); ConsumerGroupDetails getConsumerGroupDetails(final String consumerGroupId); List<ConsumerGroupDetails> getConsumerGroupDetails(final List<String> consumerGroupIds); ConsumerGroupOffsets getConsumerGroupOffsets(final String consumerGroupId); ConsumerGroupOffsetsWithTailPositions getConsumerGroupOffsetsWithTailOffsets(final String consumerGroupId); TailOffsets getTailOffsets(final String topic); TailOffsets getTailOffsets(final String topic, final Collection<Integer> partitions); void close();  }
@Test public void testCreateTopic() { final String newTopic = "TestTopic-" + System.currentTimeMillis(); TopicList topicsList = kafkaOperations.getAvailableTopics(); assertFalse("Should not contain our topic yet", topicsList.getTopicNames().contains(newTopic)); final boolean result = kafkaOperations.createTopic(new CreateTopic(newTopic, 1, (short) 1)); assertTrue("Should have true return result", result); topicsList = kafkaOperations.getAvailableTopics(); assertTrue("Should contain our topic now", topicsList.getTopicNames().contains(newTopic)); }