TopicOperator { Future<Void> onTopicCreated(LogContext logContext, TopicName topicName) { Reconciliation action = new Reconciliation("onTopicCreated") { @Override public Future<Void> execute() { Reconciliation self = this; Promise<Void> promise = Promise.promise(); TopicMetadataHandler handler = new TopicMetadataHandler(vertx, kafka, topicName, topicMetadataBackOff()) { @Override public void handle(AsyncResult<TopicMetadata> metadataResult) { if (metadataResult.succeeded()) { if (metadataResult.result() == null) { retry(); } else { Topic kafkaTopic = TopicSerialization.fromTopicMetadata(metadataResult.result()); reconcileOnTopicChange(logContext, topicName, kafkaTopic, self) .onComplete(promise); } } else { promise.fail(metadataResult.cause()); } } @Override public void onMaxAttemptsExceeded(MaxAttemptsExceededException e) { promise.fail(e); } }; kafka.topicMetadata(topicName).onComplete(handler); return promise.future(); } }; return executeWithTopicLockHeld(logContext, topicName, action); } 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 testOnTopicCreated(VertxTestContext context) { TopicMetadata topicMetadata = Utils.getTopicMetadata(topicName.toString(), new org.apache.kafka.clients.admin.Config(Collections.emptyList())); mockTopicStore.setCreateTopicResponse(topicName, null); mockKafka.setTopicMetadataResponse(topicName, topicMetadata, null); mockK8s.setCreateResponse(resourceName, null); LogContext logContext = LogContext.zkWatch(" Checkpoint async = context.checkpoint(); topicOperator.onTopicCreated(logContext, topicName).onComplete(ar -> { assertSucceeded(context, ar); mockK8s.assertExists(context, resourceName); Topic t = TopicSerialization.fromTopicMetadata(topicMetadata); mockTopicStore.assertContains(context, t); async.flag(); }); }
@Test public void testOnTopicCreated_retry(VertxTestContext context) { TopicMetadata topicMetadata = Utils.getTopicMetadata(topicName.toString(), new org.apache.kafka.clients.admin.Config(Collections.emptyList())); mockTopicStore.setCreateTopicResponse(topicName, null); AtomicInteger counter = new AtomicInteger(); mockKafka.setTopicMetadataResponse(t -> { int count = counter.getAndIncrement(); if (count == 3) { return Future.succeededFuture(topicMetadata); } else if (count < 3) { return Future.succeededFuture(null); } context.failNow(new Throwable("This should never happen")); return Future.failedFuture("This should never happen"); }); mockK8s.setCreateResponse(resourceName, null); LogContext logContext = LogContext.zkWatch(" Checkpoint async = context.checkpoint(); topicOperator.onTopicCreated(logContext, topicName).onComplete(ar -> { assertSucceeded(context, ar); context.verify(() -> assertThat(counter.get(), is(4))); mockK8s.assertExists(context, resourceName); mockTopicStore.assertContains(context, TopicSerialization.fromTopicMetadata(topicMetadata)); 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(0L)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().totalTime(TimeUnit.MILLISECONDS), is(0.0)); }); async.flag(); }); }
@Test public void testOnTopicCreated_retryTimeout(VertxTestContext context) { mockKafka.setTopicMetadataResponse(topicName, null, null); LogContext logContext = LogContext.zkWatch(" Checkpoint async = context.checkpoint(); topicOperator.onTopicCreated(logContext, topicName).onComplete(ar -> { assertFailed(context, ar); context.verify(() -> assertThat(ar.cause().getClass().getName(), is(MaxAttemptsExceededException.class.getName()))); mockK8s.assertNotExists(context, resourceName); mockTopicStore.assertNotExists(context, topicName); 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(0L)); assertThat(registry.get(TopicOperator.METRICS_PREFIX + "reconciliations.duration").tag("kind", "KafkaTopic").timer().totalTime(TimeUnit.MILLISECONDS), is(0.0)); }); async.flag(); }); }