package dev.knative.eventing.kafka.broker.core.reconciler.impl;

import static net.logstash.logback.argument.StructuredArguments.keyValue;

import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
import io.vertx.core.Handler;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;
import io.vertx.core.eventbus.MessageConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ResourcesReconcilerMessageHandler implements Handler<Message<Object>> {

  private static final Logger logger = LoggerFactory.getLogger(ResourcesReconcilerMessageHandler.class);

  public final static String ADDRESS = "resourcesreconciler.core";

  private final ResourcesReconciler resourcesReconciler;

  public ResourcesReconcilerMessageHandler(
    ResourcesReconciler resourcesReconciler) {
    this.resourcesReconciler = resourcesReconciler;
  }

  @Override
  public void handle(Message<Object> event) {
    DataPlaneContract.Contract contract = (DataPlaneContract.Contract) event.body();
    resourcesReconciler.reconcile(contract.getResourcesList())
      .onSuccess(v -> logger.info("reconciled objects {}", keyValue("contract", contract)))
      .onFailure(cause -> logger.error("failed to reconcile {}", keyValue("contract", contract), cause));
  }

  public static MessageConsumer<Object> start(EventBus eventBus, ResourcesReconciler reconciler) {
    return eventBus.localConsumer(ADDRESS, new ResourcesReconcilerMessageHandler(reconciler));
  }
}

@@ -1183,16 +1183,25 @@
 Success(
+%0A        
 v -%3E log
@@ -1225,15 +1225,27 @@
 led 
-objects
+contract generation
  %7B%7D%22
@@ -1256,32 +1256,42 @@
 yValue(%22contract
+Generation
 %22, contract)))%0A 
@@ -1285,16 +1285,32 @@
 contract
+.getGeneration()
 )))%0A    
@@ -1337,16 +1337,25 @@
 %3E logger
+%0A        
 .error(%22
@@ -1374,16 +1374,36 @@
 concile 
+contract generation 
 %7B%7D%22, key
@@ -1417,16 +1417,26 @@
 contract
+Generation
 %22, contr
@@ -1442,10 +1442,36 @@
 ract
-),
+.getGeneration()),%0A         
  cau

Fix thread blocked issue (#378)