TopicOperator { Future<Void> onResourceEvent(LogContext logContext, KafkaTopic modifiedTopic, Watcher.Action action) { return executeWithTopicLockHeld(logContext, new TopicName(modifiedTopic), new Reconciliation("onResourceEvent") { @Override public Future<Void> execute() { return k8s.getFromName(new ResourceName(modifiedTopic)) .compose(mt -> { final Topic k8sTopic; if (mt != null) { Long generation = statusUpdateGeneration.get(mt.getMetadata().getName()); LOGGER.debug("{}: last updated generation={}", logContext, generation); if (mt.getMetadata() != null && mt.getMetadata().getGeneration() != null) { if (mt.getMetadata().getGeneration().equals(generation)) { LOGGER.debug("{}: Ignoring modification event caused by my own status update on {}", logContext, mt.getMetadata().getName()); return Future.succeededFuture(); } else { LOGGER.debug("{}: modifiedTopic.getMetadata().getGeneration()={}", logContext, mt.getMetadata().getGeneration()); } } else { LOGGER.debug("{}: modifiedTopic.getMetadata().getGeneration()=null", logContext); } observedTopicFuture(mt); try { k8sTopic = TopicSerialization.fromTopicResource(mt); } catch (InvalidTopicException e) { return Future.failedFuture(e); } } else { k8sTopic = null; } return reconcileOnResourceChange(this, logContext, mt != null ? mt : modifiedTopic, k8sTopic, action == Watcher.Action.MODIFIED); }); } }); } 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 testOnKafkaTopicAdded_invalidResource(VertxTestContext context) { KafkaTopic kafkaTopic = new KafkaTopicBuilder() .withMetadata(new ObjectMetaBuilder().withName("invalid").withLabels(labels.labels()).build()) .withNewSpec() .withReplicas(1) .withPartitions(1) .withConfig(singletonMap(null, null)) .endSpec() .build(); mockK8s.setGetFromNameResponse(new ResourceName(kafkaTopic), Future.succeededFuture(kafkaTopic)); LogContext logContext = LogContext.kubeWatch(Watcher.Action.ADDED, kafkaTopic); Checkpoint async = context.checkpoint(); topicOperator.onResourceEvent(logContext, kafkaTopic, ADDED).onComplete(ar -> { assertFailed(context, ar); context.verify(() -> assertThat(ar.cause(), instanceOf(InvalidTopicException.class))); context.verify(() -> assertThat(ar.cause().getMessage(), is("KafkaTopic's spec.config has invalid entry: The key 'null' of the topic config is invalid: The value corresponding to the key must have a string, number or boolean value but the value was null"))); mockKafka.assertEmpty(context); mockTopicStore.assertEmpty(context); assertNotReadyStatus(context, new InvalidTopicException(null, ar.cause().getMessage())); 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(); }); }