TopicOperator { Future<?> reconcileAllTopics(String reconciliationType) { LOGGER.info("Starting {} reconciliation", reconciliationType); return kafka.listTopics().recover(ex -> Future.failedFuture( new OperatorException("Error listing existing topics during " + reconciliationType + " reconciliation", ex) )).compose(topicNamesFromKafka -> reconcileFromKafka(reconciliationType, topicNamesFromKafka.stream().map(TopicName::new).collect(Collectors.toList())) ).compose(reconcileState -> { Future<List<KafkaTopic>> ktFut = k8s.listResources(); return ktFut.recover(ex -> Future.failedFuture( new OperatorException("Error listing existing KafkaTopics during " + reconciliationType + " reconciliation", ex) )).map(ktList -> { reconcileState.setKafkaTopics(ktList); return reconcileState; }); }).compose(reconcileState -> { List<Future> futs = new ArrayList<>(); topicCounter.set(reconcileState.ktList.size()); for (KafkaTopic kt : reconcileState.ktList) { LogContext logContext = LogContext.periodic(reconciliationType + "kube " + kt.getMetadata().getName()).withKubeTopic(kt); Topic topic = TopicSerialization.fromTopicResource(kt); TopicName topicName = topic.getTopicName(); if (reconcileState.failed.containsKey(topicName)) { LOGGER.trace("{}: Already failed to reconcile {}", logContext, topicName); } else if (reconcileState.succeeded.contains(topicName)) { LOGGER.trace("{}: Already successfully reconciled {}", logContext, topicName); } else if (reconcileState.undetermined.contains(topicName)) { futs.add(reconcileWithKubeTopic(logContext, kt, reconciliationType, new ResourceName(kt), topic.getTopicName()).compose(r -> { reconcileState.undetermined.remove(topicName); reconcileState.succeeded.add(topicName); return Future.succeededFuture(Boolean.TRUE); })); } else { LOGGER.debug("{}: Topic {} exists in Kubernetes, but not Kafka", logContext, topicName, logTopic(kt)); futs.add(reconcileWithKubeTopic(logContext, kt, reconciliationType, new ResourceName(kt), topic.getTopicName()).compose(r -> { reconcileState.succeeded.add(topicName); return Future.succeededFuture(Boolean.TRUE); })); } } return CompositeFuture.join(futs).compose(joined -> { List<Future> futs2 = new ArrayList<>(); for (Throwable exception : reconcileState.failed.values()) { futs2.add(Future.failedFuture(exception)); } for (TopicName tn : reconcileState.undetermined) { LogContext logContext = LogContext.periodic(reconciliationType + "-" + tn); futs2.add(executeWithTopicLockHeld(logContext, tn, new Reconciliation("delete-remaining") { @Override public Future<Void> execute() { observedTopicFuture(null); return getKafkaAndReconcile(this, logContext, tn, null, null); } })); } return CompositeFuture.join(futs2); }); }); } TopicOperator(Vertx vertx, Kafka kafka,
                         K8s k8s,
                         TopicStore topicStore,
                         Labels labels,
                         String namespace,
                         Config config,
                         MetricsProvider metrics); void initMetrics(); Counter getPeriodicReconciliationsCounter(); void setTopicCount(int topics); Future<Void> executeWithTopicLockHeld(LogContext logContext, TopicName key, Reconciliation action); boolean isWorkInflight(); static final String METRICS_PREFIX; }
@Test public void testReconcileAllTopics_listTopicsFails(VertxTestContext context) { RuntimeException error = new RuntimeException("some failure"); mockKafka.setTopicsListResponse(Future.failedFuture(error)); Future<?> reconcileFuture = topicOperator.reconcileAllTopics("periodic"); reconcileFuture.onComplete(context.failing(e -> { context.verify(() -> assertThat(e.getMessage(), is("Error listing existing topics during periodic reconciliation"))); context.verify(() -> assertThat(e.getCause(), is(error))); context.completeNow(); })); }
@Test public void testReconcileAllTopics_getResourceFails(VertxTestContext context) { RuntimeException error = new RuntimeException("some failure"); mockKafka.setTopicsListResponse(Future.succeededFuture(singleton(topicName.toString()))); mockKafka.setDeleteTopicResponse(topicName, null); mockTopicStore.setGetTopicResponse(topicName, Future.failedFuture(error)); Future<?> reconcileFuture = topicOperator.reconcileAllTopics("periodic"); reconcileFuture.onComplete(context.failing(e -> { context.verify(() -> assertThat(e.getMessage(), is("Error getting topic my-topic from topic store during periodic reconciliation"))); context.verify(() -> assertThat(e.getCause(), is(error))); context.verify(() -> { MeterRegistry registry = metrics.meterRegistry(); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations").tag("kind", "KafkaTopic").counter().count(), is(2.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.successful").tag("kind", "KafkaTopic").counter().count(), is(0.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.failed").tag("kind", "KafkaTopic").counter().count(), is(1.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().count(), is(1L)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().totalTime(TimeUnit.MILLISECONDS), greaterThan(0.0)); }); context.completeNow(); })); }
@Test public void testReconcileAllTopics_listMapsFails(VertxTestContext context) { RuntimeException error = new RuntimeException("some failure"); mockKafka.setTopicsListResponse(Future.succeededFuture(emptySet())); mockK8s.setListMapsResult(() -> Future.failedFuture(error)); Future<?> reconcileFuture = topicOperator.reconcileAllTopics("periodic"); reconcileFuture.onComplete(context.failing(e -> { context.verify(() -> assertThat(e.getMessage(), is("Error listing existing KafkaTopics during periodic reconciliation"))); context.verify(() -> assertThat(e.getCause(), is(error))); context.completeNow(); })); }
@Test public void testReconcileMetrics(VertxTestContext context) throws InterruptedException { mockKafka.setTopicsListResponse(Future.succeededFuture(emptySet())); mockKafka.setUpdateTopicResponse(topicName -> Future.succeededFuture()); Future<?> reconcileFuture = topicOperator.reconcileAllTopics("periodic"); resourceAdded(context, null, null); Checkpoint async = context.checkpoint(); reconcileFuture.onComplete(context.succeeding(e -> context.verify(() -> { MeterRegistry registry = metrics.meterRegistry(); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations").tag("kind", "KafkaTopic").counter().count(), is(1.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.successful").tag("kind", "KafkaTopic").counter().count(), is(0.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.failed").tag("kind", "KafkaTopic").counter().count(), is(0.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().count(), is(1L)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().totalTime(TimeUnit.MILLISECONDS), greaterThan(0.0)); async.flag(); }))); }
@Test public void testReconcileMetricsDeletedTopic(VertxTestContext context) throws InterruptedException { mockKafka.setTopicsListResponse(Future.succeededFuture(emptySet())); mockKafka.setUpdateTopicResponse(topicName -> Future.succeededFuture()); Future<?> reconcileFuture = topicOperator.reconcileAllTopics("periodic"); resourceAdded(context, null, null); resourceRemoved(context, null, null); Checkpoint async = context.checkpoint(); reconcileFuture.onComplete(context.succeeding(e -> context.verify(() -> { MeterRegistry registry = metrics.meterRegistry(); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations").tag("kind", "KafkaTopic").counter().count(), is(0.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.successful").tag("kind", "KafkaTopic").counter().count(), is(0.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.failed").tag("kind", "KafkaTopic").counter().count(), is(0.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().count(), is(1L)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().totalTime(TimeUnit.MILLISECONDS), greaterThan(0.0)); async.flag(); }))); }