Filter unknown process definitions received via kafka #74

Author: martingreilinger

adapters/camunda8/src/main/java/io/vanillabp/cockpit/adapter/camunda8/receiver/kafka/KafkaController.java

@@ -1,31 +1,27 @@
 package io.vanillabp.cockpit.adapter.camunda8.receiver.kafka;
 
-
 import at.phactum.zeebe.exporters.kafka.serde.RecordId;
 import io.camunda.zeebe.protocol.record.Record;
 import io.camunda.zeebe.protocol.record.RecordType;
 import io.camunda.zeebe.protocol.record.ValueType;
 import io.camunda.zeebe.protocol.record.intent.ProcessEventIntent;
 import io.camunda.zeebe.protocol.record.intent.ProcessInstanceIntent;
-import io.camunda.zeebe.protocol.record.value.BpmnElementType;
-import io.camunda.zeebe.protocol.record.value.JobRecordValue;
-import io.camunda.zeebe.protocol.record.value.ProcessEventRecordValue;
-import io.camunda.zeebe.protocol.record.value.ProcessInstanceCreationRecordValue;
-import io.camunda.zeebe.protocol.record.value.ProcessInstanceRecordValue;
+import io.camunda.zeebe.protocol.record.value.*;
 import io.vanillabp.cockpit.adapter.camunda8.deployments.DeploymentService;
 import io.vanillabp.cockpit.adapter.camunda8.receiver.events.Camunda8UserTaskCreatedEvent;
 import io.vanillabp.cockpit.adapter.camunda8.receiver.events.Camunda8UserTaskLifecycleEvent;
 import io.vanillabp.cockpit.adapter.camunda8.receiver.events.Camunda8WorkflowLifeCycleEvent;
 import io.vanillabp.cockpit.adapter.camunda8.usertask.Camunda8UserTaskEventHandler;
 import io.vanillabp.cockpit.adapter.camunda8.workflow.Camunda8WorkflowEventHandler;
 import io.vanillabp.springboot.adapter.VanillaBpProperties;
-import java.util.Set;
-import java.util.function.Supplier;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.kafka.annotation.KafkaListener;
 
+import java.util.Set;
+import java.util.function.Supplier;
+
 import static io.vanillabp.cockpit.adapter.camunda8.receiver.kafka.KafkaConfiguration.KAFKA_CONSUMER_PREFIX;
 
 public class KafkaController {
@@ -85,6 +81,15 @@ private void handleProcessInstanceCreationRecord(
         }
         // empty (none) start event
         final var processInstanceCreationRecordValue = (ProcessInstanceCreationRecordValue) value.getValue();
+	final var processDefinitionKey = processInstanceCreationRecordValue.getProcessDefinitionKey();
+	if (hasNoProcessInformation(processDefinitionKey)) {
+	    final var tenantId = processInstanceCreationRecordValue.getTenantId();
+	    if (camunda8WorkflowEventHandler.isTenantKnown(tenantId)) {
+		logUnknownProcessDefinitionWarning(tenantId, processDefinitionKey);
+	    }
+	    return;
+	}
+
         final var workflowCreatedEvent = WorkflowEventZeebeRecordMapper.map(processInstanceCreationRecordValue);
         WorkflowEventZeebeRecordMapper.addMetaData(workflowCreatedEvent, value, idNames);
         camunda8WorkflowEventHandler.saveBusinessKeyForRootProcessInstance(workflowCreatedEvent);
@@ -171,6 +176,16 @@ private void handleJobRecord(Record<?> value) {
         if (!jobRecordValue.getType().equals("io.camunda.zeebe:userTask")){
             return;
         }
+
+	long processDefinitionKey = jobRecordValue.getProcessDefinitionKey();
+	if (hasNoProcessInformation(processDefinitionKey)) {
+	    String tenantId = jobRecordValue.getTenantId();
+	    if (camunda8WorkflowEventHandler.isTenantKnown(tenantId)) {
+		logUnknownProcessDefinitionWarning(tenantId, processDefinitionKey);
+	    }
+	    return;
+	}
+
         // job lifecycle
         String intentName = value.getIntent().name();
         if (intentName.equals("CREATED")) {
@@ -187,4 +202,13 @@ private void handleJobRecord(Record<?> value) {
 
     }
 
+    private boolean hasNoProcessInformation(long definitionKey) {
+	return deploymentService.getProcessInformationByDefinitionKey(definitionKey).isEmpty();
+    }
+
+    private void logUnknownProcessDefinitionWarning(String tenantId, long processDefinitionKey) {
+	logger.warn("Tenant {} has no process information for definition key '{}'!",
+		tenantId,
+		processDefinitionKey);
+    }
 }