ZkTopicsWatcher { void start(Zk zk) { children = null; tcw.start(zk); tw.start(zk); zk.watchChildren(TOPICS_ZNODE, childResult -> { if (state == 2) { zk.unwatchChildren(TOPICS_ZNODE); return; } if (childResult.failed()) { LOGGER.error("Error on znode {} children", TOPICS_ZNODE, childResult.cause()); return; } List<String> result = childResult.result(); LOGGER.debug("znode {} now has children {}, previous children {}", TOPICS_ZNODE, result, this.children); Set<String> deleted = new HashSet<>(this.children); deleted.removeAll(result); Set<String> created = new HashSet<>(result); created.removeAll(this.children); this.children = result; if (!deleted.isEmpty()) { LOGGER.info("Deleted topics: {}", deleted); for (String topicName : deleted) { tcw.removeChild(topicName); tw.removeChild(topicName); LogContext logContext = LogContext.zkWatch(TOPICS_ZNODE, "-" + topicName); topicOperator.onTopicDeleted(logContext, new TopicName(topicName)).onComplete(ar -> { if (ar.succeeded()) { LOGGER.debug("{}: Success responding to deletion of topic {}", logContext, topicName); } else { LOGGER.warn("{}: Error responding to deletion of topic {}", logContext, topicName, ar.cause()); } }); } } if (!created.isEmpty()) { LOGGER.info("Created topics: {}", created); for (String topicName : created) { tcw.addChild(topicName); tw.addChild(topicName); LogContext logContext = LogContext.zkWatch(TOPICS_ZNODE, "+" + topicName); topicOperator.onTopicCreated(logContext, new TopicName(topicName)).onComplete(ar -> { if (ar.succeeded()) { LOGGER.debug("{}: Success responding to creation of topic {}", logContext, topicName); } else { LOGGER.warn("{}: Error responding to creation of topic {}", logContext, topicName, ar.cause()); } }); } } }).<Void>compose(zk2 -> { zk.children(TOPICS_ZNODE, childResult -> { if (childResult.failed()) { LOGGER.error("Error on znode {} children", TOPICS_ZNODE, childResult.cause()); return; } List<String> result = childResult.result(); LOGGER.debug("Setting initial children {}", result); this.children = result; for (String child : result) { tcw.addChild(child); tw.addChild(child); } this.state = 1; }); return Future.succeededFuture(); }); } ZkTopicsWatcher(TopicOperator topicOperator, TopicConfigsWatcher tcw, ZkTopicWatcher tw);   }
@Test public void testTopicDelete() { operator.topicDeletedResult = Future.succeededFuture(); mockZk.childrenResult = Future.succeededFuture(asList("foo", "bar")); TopicConfigsWatcher topicConfigsWatcher = new TopicConfigsWatcher(operator); ZkTopicWatcher topicWatcher = new ZkTopicWatcher(operator); ZkTopicsWatcher topicsWatcher = new ZkTopicsWatcher(operator, topicConfigsWatcher, topicWatcher); topicsWatcher.start(mockZk); mockZk.triggerChildren(Future.succeededFuture(asList("foo"))); assertThat(operator.getMockOperatorEvents(), is(asList(new MockTopicOperator.MockOperatorEvent( Type.DELETE, new TopicName("bar"))))); assertThat(topicConfigsWatcher.watching("baz"), is(false)); }