TopicOperator { Future<Void> reconcile(Reconciliation reconciliation, final LogContext logContext, final HasMetadata involvedObject, final Topic k8sTopic, final Topic kafkaTopic, final Topic privateTopic) { final Future<Void> reconciliationResultHandler; { TopicName topicName = k8sTopic != null ? k8sTopic.getTopicName() : kafkaTopic != null ? kafkaTopic.getTopicName() : privateTopic != null ? privateTopic.getTopicName() : null; LOGGER.info("{}: Reconciling topic {}, k8sTopic:{}, kafkaTopic:{}, privateTopic:{}", logContext, topicName, k8sTopic == null ? "null" : "nonnull", kafkaTopic == null ? "null" : "nonnull", privateTopic == null ? "null" : "nonnull"); } if (privateTopic == null) { if (k8sTopic == null) { if (kafkaTopic == null) { LOGGER.debug("{}: All three topics null during reconciliation.", logContext); reconciliationResultHandler = Future.succeededFuture(); } else { LOGGER.debug("{}: topic created in kafka, will create KafkaTopic in k8s and topicStore", logContext); reconciliationResultHandler = createResource(logContext, kafkaTopic) .compose(createdKt -> { reconciliation.observedTopicFuture(createdKt); return createInTopicStore(logContext, kafkaTopic, involvedObject); }); } } else if (kafkaTopic == null) { LOGGER.debug("{}: KafkaTopic created in k8s, will create topic in kafka and topicStore", logContext); reconciliationResultHandler = createKafkaTopic(logContext, k8sTopic, involvedObject) .compose(ignore -> createInTopicStore(logContext, k8sTopic, involvedObject)) .compose(ignored -> getFromKafka(k8sTopic.getTopicName())) .compose(kafkaTopic2 -> { LOGGER.debug("Post-create kafka {}", kafkaTopic2); if (kafkaTopic2 == null) { LOGGER.error("Post-create kafka unexpectedly null"); return Future.succeededFuture(); } return update3Way(reconciliation, logContext, involvedObject, k8sTopic, kafkaTopic2, k8sTopic); }); } else { reconciliationResultHandler = update2Way(reconciliation, logContext, involvedObject, k8sTopic, kafkaTopic); } } else { if (k8sTopic == null) { if (kafkaTopic == null) { LOGGER.debug("{}: KafkaTopic deleted in k8s and topic deleted in kafka => delete from topicStore", logContext); reconciliationResultHandler = deleteFromTopicStore(logContext, involvedObject, privateTopic.getTopicName()); } else { LOGGER.debug("{}: KafkaTopic deleted in k8s => delete topic from kafka and from topicStore", logContext); reconciliationResultHandler = deleteKafkaTopic(logContext, kafkaTopic.getTopicName()).recover(this::handleTopicDeletionDisabled) .compose(ignored -> deleteFromTopicStore(logContext, involvedObject, privateTopic.getTopicName())); } } else if (kafkaTopic == null) { LOGGER.debug("{}: topic deleted in kafkas => delete KafkaTopic from k8s and from topicStore", logContext); reconciliationResultHandler = deleteResource(logContext, privateTopic.getOrAsKubeName()) .compose(ignore -> { reconciliation.observedTopicFuture(null); return deleteFromTopicStore(logContext, involvedObject, privateTopic.getTopicName()); }); } else { LOGGER.debug("{}: 3 way diff", logContext); reconciliationResultHandler = update3Way(reconciliation, logContext, involvedObject, k8sTopic, kafkaTopic, privateTopic); } } return reconciliationResultHandler; } 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 testReconcile_noResource_withKafka_noPrivate(VertxTestContext context) throws InterruptedException { Topic kubeTopic = null; Topic kafkaTopic = new Topic.Builder(topicName.toString(), 10, (short) 2, map("cleanup.policy", "bar"), metadata).build(); Topic privateTopic = null; CountDownLatch async0 = new CountDownLatch(1); mockTopicStore.setCreateTopicResponse(topicName, null); mockK8s.setCreateResponse(topicName.asKubeName(), null); mockKafka.setCreateTopicResponse(topicName -> Future.succeededFuture()); mockKafka.createTopic(kafkaTopic).onComplete(ar -> async0.countDown()); async0.await(); LogContext logContext = LogContext.periodic(topicName.toString()); CountDownLatch async = new CountDownLatch(2); topicOperator.reconcile(reconciliation(), logContext, null, kubeTopic, kafkaTopic, privateTopic).onComplete(reconcileResult -> { assertSucceeded(context, reconcileResult); mockTopicStore.assertExists(context, topicName); mockK8s.assertExists(context, topicName.asKubeName()); mockKafka.assertExists(context, topicName); mockK8s.assertNoEvents(context); mockTopicStore.read(topicName).onComplete(readResult -> { assertSucceeded(context, readResult); context.verify(() -> assertThat(readResult.result(), is(kafkaTopic))); async.countDown(); }); mockK8s.getFromName(topicName.asKubeName()).onComplete(readResult -> { assertSucceeded(context, readResult); context.verify(() -> assertThat(TopicSerialization.fromTopicResource(readResult.result()), is(kafkaTopic))); async.countDown(); }); try { async.await(60, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } context.verify(() -> assertThat(mockKafka.getTopicState(topicName), is(kafkaTopic))); 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.locked").tag("kind", "KafkaTopic").counter().count(), is(0.0)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().count(), is(0L)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().totalTime(TimeUnit.MILLISECONDS), is(0.0)); }); context.completeNow(); }); }
@Test public void testReconcile_noResource_withKafka_withPrivate(VertxTestContext context) throws InterruptedException { Topic kubeTopic = null; Topic kafkaTopic = new Topic.Builder(topicName.toString(), 10, (short) 2, map("cleanup.policy", "bar")).build(); Topic privateTopic = kafkaTopic; CountDownLatch async0 = new CountDownLatch(2); mockKafka.createTopic(kafkaTopic).onComplete(ar -> async0.countDown()); mockKafka.setDeleteTopicResponse(topicName, null); mockTopicStore.setCreateTopicResponse(topicName, null); mockTopicStore.create(kafkaTopic).onComplete(ar -> async0.countDown()); mockTopicStore.setDeleteTopicResponse(topicName, null); async0.await(); LogContext logContext = LogContext.periodic(topicName.toString()); Checkpoint async = context.checkpoint(); topicOperator.reconcile(reconciliation(), logContext, null, kubeTopic, kafkaTopic, privateTopic).onComplete(reconcileResult -> { assertSucceeded(context, reconcileResult); mockTopicStore.assertNotExists(context, topicName); mockK8s.assertNotExists(context, topicName.asKubeName()); mockKafka.assertNotExists(context, topicName); mockK8s.assertNoEvents(context); async.flag(); }); }