KafkaImpl implements Kafka { @Override public Future<TopicMetadata> topicMetadata(TopicName topicName) { LOGGER.debug("Getting metadata for topic {}", topicName); ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topicName.toString()); Future<TopicDescription> topicDescriptionFuture = mapFuture(adminClient.describeTopics( singleton(topicName.toString())).values().get(topicName.toString())); Future<Config> configFuture = mapFuture(adminClient.describeConfigs( singleton(resource)).values().get(resource)); return CompositeFuture.all(topicDescriptionFuture, configFuture) .map(compositeFuture -> new TopicMetadata(compositeFuture.resultAt(0), compositeFuture.resultAt(1))) .recover(error -> { if (error instanceof UnknownTopicOrPartitionException) { return Future.succeededFuture(null); } else { return Future.failedFuture(error); } }); } KafkaImpl(Admin adminClient, Vertx vertx); @Override Future<Void> deleteTopic(TopicName topicName); @SuppressWarnings("deprecation") @Override Future<Void> updateTopicConfig(Topic topic); @Override Future<TopicMetadata> topicMetadata(TopicName topicName); @Override Future<Set<String>> listTopics(); @Override Future<Void> increasePartitions(Topic topic); @Override Future<Void> createTopic(Topic topic);  }
@Test public void testTopicMetadataBothNotFound(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDescribeTopics(admin, singletonMap("test", Either.ofRight(new UnknownTopicOrPartitionException()))); mockDescribeConfigs(admin, singletonMap( new ConfigResource(ConfigResource.Type.TOPIC, "test"), Either.ofRight(new UnknownTopicOrPartitionException()))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.topicMetadata(new TopicName("test")).onComplete(testContext.succeeding(topicMetadata -> testContext.verify(() -> { assertNull(topicMetadata); testContext.completeNow(); }))); }
@Test public void testTopicMetadataDescribeTopicNotFound(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDescribeTopics(admin, singletonMap("test", Either.ofRight(new UnknownTopicOrPartitionException()))); mockDescribeConfigs(admin, singletonMap( new ConfigResource(ConfigResource.Type.TOPIC, "test"), Either.ofLeft(mock(Config.class)))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.topicMetadata(new TopicName("test")).onComplete(testContext.succeeding(topicMetadata -> testContext.verify(() -> { assertNull(topicMetadata); testContext.completeNow(); }))); }
@Test public void testTopicMetadataDescribeConfigsNotFound(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDescribeTopics(admin, singletonMap("test", Either.ofLeft(mock(TopicDescription.class)))); mockDescribeConfigs(admin, singletonMap( new ConfigResource(ConfigResource.Type.TOPIC, "test"), Either.ofRight(new UnknownTopicOrPartitionException()))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.topicMetadata(new TopicName("test")).onComplete(testContext.succeeding(topicMetadata -> testContext.verify(() -> { assertNull(topicMetadata); testContext.completeNow(); }))); }
@Test public void testTopicMetadataBothFound(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDescribeTopics(admin, singletonMap("test", Either.ofLeft(mock(TopicDescription.class)))); mockDescribeConfigs(admin, singletonMap(new ConfigResource(ConfigResource.Type.TOPIC, "test"), Either.ofLeft(mock(Config.class)))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.topicMetadata(new TopicName("test")).onComplete(testContext.succeeding(topicMetadata -> testContext.verify(() -> { assertNotNull(topicMetadata); assertNotNull(topicMetadata.getDescription()); assertNotNull(topicMetadata.getConfig()); testContext.completeNow(); }))); }
@Test public void testTopicMetadataDescribeTimeout(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDescribeTopics(admin, singletonMap("test", Either.ofLeft(mock(TopicDescription.class)))); mockDescribeConfigs(admin, singletonMap(new ConfigResource(ConfigResource.Type.TOPIC, "test"), Either.ofRight(new TimeoutException()))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.topicMetadata(new TopicName("test")).onComplete(testContext.failing(error -> testContext.verify(() -> { assertTrue(error instanceof TimeoutException); testContext.completeNow(); }))); }