KafkaImpl implements Kafka { @Override public Future<Void> deleteTopic(TopicName topicName) { Promise<Void> handler = Promise.promise(); LOGGER.debug("Deleting topic {}", topicName); KafkaFuture<Void> future = adminClient.deleteTopics( singleton(topicName.toString())).values().get(topicName.toString()); mapFuture(future).compose(ig -> Util.waitFor(vertx, "deleted sync " + topicName, "deleted", 1000, 120_000, () -> { try { return adminClient.describeTopics(singleton(topicName.toString())).all().get().get(topicName.toString()) == null; } catch (ExecutionException e) { if (e.getCause() instanceof UnknownTopicOrPartitionException) { return true; } else if (e.getCause() instanceof RuntimeException) { throw (RuntimeException) e.getCause(); } else { throw new RuntimeException(e); } } catch (InterruptedException e) { throw new RuntimeException(e); } }, error -> true) ).onComplete(ar -> { vertx.runOnContext(ignored -> { handler.handle(ar); }); }); return handler.future(); } 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 testDelete(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDescribeTopics(admin, singletonMap("test", Either.ofRight(new UnknownTopicOrPartitionException()))); mockDeleteTopics(admin, singletonMap("test", Either.ofLeft(null))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.deleteTopic(new TopicName("test")).onComplete(testContext.succeeding(error -> testContext.verify(() -> { testContext.completeNow(); }))); }
@Test public void testDeleteDeleteTimeout(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDescribeTopics(admin, singletonMap("test", Either.ofRight(new UnknownTopicOrPartitionException()))); mockDeleteTopics(admin, singletonMap("test", Either.ofRight(new TimeoutException()))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.deleteTopic(new TopicName("test")).onComplete(testContext.failing(error -> testContext.verify(() -> { assertTrue(error instanceof TimeoutException); testContext.completeNow(); }))); }
@Test public void testDeleteDescribeTimeout(VertxTestContext testContext) { Admin admin = mock(Admin.class); mockDeleteTopics(admin, singletonMap("test", Either.ofLeft(null))); mockDescribeTopics(admin, singletonMap("test", Either.ofRight(new TimeoutException()))); KafkaImpl impl = new KafkaImpl(admin, vertx); impl.deleteTopic(new TopicName("test")).onComplete(testContext.failing(error -> testContext.verify(() -> { assertTrue(error instanceof TimeoutException); testContext.completeNow(); }))); }