Problems executing a connector

Juan_Felipe_Parrado · October 9, 2023, 10:35pm

I have created a very simple process:

<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" xmlns:zeebe="http://camunda.org/schema/zeebe/1.0" xmlns:modeler="http://camunda.org/schema/modeler/1.0" xmlns:camunda="http://camunda.org/schema/1.0/bpmn" id="Definitions_1" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Web Modeler" exporterVersion="6c48892" modeler:executionPlatform="Camunda Cloud" modeler:executionPlatformVersion="8.3.0" camunda:diagramRelationId="cd5250e0-e339-43ed-9703-34e6607f0691">
  <bpmn:process id="idSimpleProces" name="SimpleProcess" isExecutable="true">
    <bpmn:startEvent id="StartEvent_1" name="start">
      <bpmn:outgoing>Flow_17jqy6e</bpmn:outgoing>
    </bpmn:startEvent>
    <bpmn:sequenceFlow id="Flow_17jqy6e" sourceRef="StartEvent_1" targetRef="idSimpleProcess" />
    <bpmn:endEvent id="Event_13eupph" name="end">
      <bpmn:incoming>Flow_05sdfes</bpmn:incoming>
    </bpmn:endEvent>
    <bpmn:task id="idSimpleProcess" name="llama endpoint" zeebe:modelerTemplate="0152a825-64ed-4a13-9987-3901de791f57" zeebe:modelerTemplateVersion="1696859318446">
      <bpmn:extensionElements>
        <zeebe:taskDefinition type="io.camunda:http-json:1" />
        <zeebe:ioMapping>
          <zeebe:input source="=inputs" target="inputs" />
          <zeebe:input source="=model_id" target="model_id" />
        </zeebe:ioMapping>
        <zeebe:taskHeaders>
          <zeebe:header key="resultVariable" value="resultVariable" />
        </zeebe:taskHeaders>
      </bpmn:extensionElements>
      <bpmn:incoming>Flow_17jqy6e</bpmn:incoming>
      <bpmn:outgoing>Flow_05sdfes</bpmn:outgoing>
    </bpmn:task>
    <bpmn:sequenceFlow id="Flow_05sdfes" sourceRef="idSimpleProcess" targetRef="Event_13eupph" />
  </bpmn:process>
  <bpmndi:BPMNDiagram id="BPMNDiagram_1">
    <bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="idSimpleProces">
      <bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
        <dc:Bounds x="172" y="100" width="36" height="36" />
        <bpmndi:BPMNLabel>
          <dc:Bounds x="179" y="143" width="23" height="14" />
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape id="Event_13eupph_di" bpmnElement="Event_13eupph">
        <dc:Bounds x="452" y="100" width="36" height="36" />
        <bpmndi:BPMNLabel>
          <dc:Bounds x="461" y="143" width="19" height="14" />
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNShape>
      <bpmndi:BPMNShape id="Activity_1tevrm6_di" bpmnElement="idSimpleProcess">
        <dc:Bounds x="290" y="78" width="100" height="80" />
        <bpmndi:BPMNLabel />
      </bpmndi:BPMNShape>
      <bpmndi:BPMNEdge id="Flow_17jqy6e_di" bpmnElement="Flow_17jqy6e">
        <di:waypoint x="208" y="118" />
        <di:waypoint x="290" y="118" />
        <bpmndi:BPMNLabel>
          <dc:Bounds x="202" y="100" width="23" height="14" />
        </bpmndi:BPMNLabel>
      </bpmndi:BPMNEdge>
      <bpmndi:BPMNEdge id="Flow_05sdfes_di" bpmnElement="Flow_05sdfes">
        <di:waypoint x="390" y="118" />
        <di:waypoint x="452" y="118" />
      </bpmndi:BPMNEdge>
    </bpmndi:BPMNPlane>
  </bpmndi:BPMNDiagram>
</bpmn:definitions>

I created the three files to manage the connector "//How to build a Camunda 8 Connector | Camunda
package io.camunda.connector;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.co.igg.catastro.bpmn.services.dto.MyConnectorRequest;
import com.co.igg.catastro.bpmn.services.dto.MyConnectorResult;

import io.camunda.connector.api.annotation.OutboundConnector;
import io.camunda.connector.api.outbound.OutboundConnectorContext;
import io.camunda.connector.api.outbound.OutboundConnectorFunction;

import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.io.BufferedReader;
import java.io.IOException;

@OutboundConnector(
		name = "base 64", 
		inputVariables = { "inputs", "model_id" }, 
		type = "io.camunda:http-json:1")
    public class MyConnectorFunction implements OutboundConnectorFunction {

        private static final Logger LOGGER = LoggerFactory.getLogger(MyConnectorFunction.class);

        @Override
        public Object execute(OutboundConnectorContext context) throws Exception {
            // aca traemos los valores de las variables y los secretos
   
            var connectorRequest = context.bindVariables(MyConnectorRequest.class);
            //context.replaceSecrets(connectorRequest);
            return executeConnector(connectorRequest);
        }

        private MyConnectorResult executeConnector(final MyConnectorRequest connectorRequest) throws IOException {
            // TODO: implement connector logic

            String urlString = "https://api.pruebas.isipoint.co:8093";
            URL url = new URL(urlString);
            HttpURLConnection http = (HttpURLConnection)url.openConnection();
            http.setRequestProperty("Accept", "application/json");

            http.disconnect();

            String documentInformation;
            if (http.getResponseCode() == 200) {
              documentInformation= convertInputStreamToString(http.getInputStream());
              LOGGER.info("Document information" + documentInformation);
            } else {
                LOGGER.error("Error accessing OpenWeather API: " + http.getResponseCode() + " - " + http.getResponseMessage());
                // Throwing an exception will fail the job
                throw new IOException(http.getResponseMessage());   
            }

            var result = new MyConnectorResult();
            result.setResult("{\"code\": " + http.getResponseCode() + ", \"documentInformation\": " + documentInformation + "}");
            
            return result;
        }
        private static String convertInputStreamToString(InputStream inputStream) {
            StringBuilder result = new StringBuilder();

            try (BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
              String line;
              while ((line = reader.readLine()) != null) {
                result.append(line);
              }
            } catch (IOException ex) {
              LOGGER.error("Error during response reading: ", ex);
              return "{}";
            }

            return result.toString();
          }
    }

"package com.co.igg.catastro.bpmn.services.dto;

public class MyConnectorRequest {
    private String inputs;
    private String model_id;

    public String getInputs() {
        return inputs;
    }

    public void setInputs(String inputs) {
        this.inputs = inputs;
    }

    public String getModel_id() {
        return model_id;
    }

    public void setModel_id(String model_id) {
        this.model_id = model_id;
    }
}```
and "package com.co.igg.catastro.bpmn.services.dto;
```java
public class MyConnectorResult {
	
	private String result;
	
	public MyConnectorResult() {
		// TODO Auto-generated constructor stub
	}
	public MyConnectorResult(String result) {
		super();
		this.result = result;
	}
	
	@Override
	public String toString() {
		return "MyConnectorResult [result=" + result + "]";
	}
	public String getResult() {
		return result;
	}
	public void setResult(String result) {
		this.result = result;
	}
}
" .

also the connector is created with a template

"{
  "$schema": "https://unpkg.com/@camunda/zeebe-element-templates-json-schema/resources/schema.json",
  "name": "base 64",
  "id": "0152a825-64ed-4a13-9987-3901de791f57",
  "appliesTo": [
    "bpmn:Task"
  ],
  "properties": [
    {
      "type": "Hidden",
      "value": "io.camunda:http-json:1",
      "binding": {
        "type": "zeebe:taskDefinition:type"
      }
    },
    {
      "label": "inputs",
      "description": "Payload to send with the request which contains an image in b64",
      "type": "String",
      "value": "=inputs",
      "feel": "optional",
      "binding": {
        "type": "zeebe:input",
        "name": "inputs"
      },
      "constraints": {
        "notEmpty": true
      }
    },
    {
      "label": "model_id",
      "description": "Payload to send with the request describing the pype of model",
      "type": "String",
      "value": "=model_id",
      "feel": "optional",
      "binding": {
        "type": "zeebe:input",
        "name": "model_id"
      },
      "constraints": {
        "notEmpty": true
      }
    },
    {
      "label": "Result Variable",
      "description": "Name of variable to store the response in",
      "group": "output",
      "type": "String",
      "value": "resultVariable",
      "binding": {
        "type": "zeebe:taskHeader",
        "key": "resultVariable"
      }
    }
  ]
}

these files are inside of my project that consist of an enpoint that starts a new instanceof the process, starting it with the 2 variables that are going to be needed in he connector. when i check, the procss ran all until the end but it does not do what is inside of execute (i putted some console logs and break points to know if it goes inside of it). so it goes through it but without doing what is inside of the code.
so i have main questions.

How can i inside of the connector get the variables that i set up at the beggining of the process(when i started it. i started it with a value)
How can i make sure that the process get inside of my @OutboundConnector ?

nathan.loding · October 11, 2023, 7:24pm

Hi @Juan_Felipe_Parrado - it sounds like you may be missing the Connector Runtime. The runtime is the bridge between your Connector code and the workflow engine.

If you have the Runtime running, my next thought is that you have duplicated an existing zeebe:taskDefinition:type - in this case, your template is using io.camunda:http-json:1 as the type, but this value is already the type for the REST Protocol Connector. Try changing that value to something unique for your Connector/namespace in both the template and your code.

Hopefully one of the two resolves the issue!

Juan_Felipe_Parrado · October 12, 2023, 8:00pm

i have added the first dependency

io.camunda.connector
spring-boot-starter-camunda-connectors
${version.connectors}

But the second one i wasnt able to, because i couldnt find the artifact or its versions anywhere.

org.myorg
connector-my-awesome
${version.connector-my-awesome}

I have a main function that wraps with a connector job handler my class
package com.co.igg.catastro.bpmn.services;

import javax.transaction.Transactional;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import com.co.igg.catastro.bpmn.interfaces.IProcessService;
import com.co.igg.catastro.bpmn.services.dto.MyConnectorFunction;
import com.co.igg.catastro.bpmn.services.dto.MyConnectorRequest;
import com.co.igg.catastro.bpmn.services.dto.MyConnectorResult;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.DeserializationFeature;
import io.camunda.zeebe.client.ZeebeClient;
import io.camunda.zeebe.client.api.response.ProcessInstanceEvent;
import io.camunda.zeebe.client.api.response.ActivatedJob;
import io.camunda.zeebe.client.api.worker.JobClient;
import io.camunda.zeebe.client.api.worker.JobHandler;
import io.camunda.zeebe.client.api.worker.JobWorker;

import org.springframework.http.HttpStatus;
import org.camunda.bpm.application.ProcessApplication;
import org.camunda.bpm.engine.ProcessEngine;
import org.camunda.bpm.engine.ProcessEngineConfiguration;
import org.camunda.bpm.engine.runtime.ActivityInstance;
import org.camunda.bpm.engine.runtime.ProcessInstance;
import org.camunda.bpm.engine.TaskService;
import org.camunda.bpm.engine.task.Task;
import org.camunda.bpm.engine.task.TaskQuery;
import org.camunda.connect.ConnectorException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.RequestBody;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;

import java.time.Duration;

import io.camunda.connector.api.annotation.OutboundConnector;

import io.camunda.connector.api.outbound.OutboundConnectorContext;
import io.camunda.connector.api.outbound.OutboundConnectorFunction;
import io.camunda.connector.runtime.util.outbound.ConnectorJobHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Service
public class ProcessService implements IProcessService {

private final Logger logger = LoggerFactory.getLogger(getClass());
private MyConnectorRequest variables = new MyConnectorRequest();

@Autowired
private ZeebeClient client;

@Override
@Transactional
public String createBpmnProcess(String processDefinitionKey, String body) {
    System.out.println("Entrando en createBpmnProcess");
    try {
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        MyConnectorRequest variables = objectMapper.readValue(body, MyConnectorRequest.class);
        final ProcessInstanceEvent processInstanceEvent = client
                .newCreateInstanceCommand()
                .bpmnProcessId(processDefinitionKey)
                .latestVersion()
                .variables(variables)
                .send()
                .join();
        Long idProceso = processInstanceEvent.getProcessInstanceKey();
        System.out.println("Deployment successful. Workflow deployed with key:"+idProceso);
        logger.info("Deployment successful. Workflow deployed with key: {}", idProceso);
        System.out.println("Saliendo de createBpmnProcess");

        client.newWorker()
                .jobType("io.camunda:unique-id-catastro")
                .handler(new ConnectorJobHandler(new MyConnectorFunction()))
                .maxJobsActive(5)
                .name("base 64")
                .pollInterval(Duration.ofMillis(100))
                .open();
        // client.close();
        return String.valueOf(idProceso);
    } catch (Exception e) {
        logger.error("Error: {}", e.toString());
        System.out.println("Saliendo de createBpmnProcess");
        // client.close();
        return "error";
    }
}


@Override
public void run(String... args) throws Exception {
    // TODO Auto-generated method stub

}

}
i have also changed the io.camunda:http-json:1 for “io.camunda:unique-id-catastro” . when i try to run a process i does go to the end but as i am telling you it doesnt get inside of my executor

nathan.loding · October 12, 2023, 8:58pm

Are you running the Connector Runtime? (See the link I provided, or the section in the blog post you linked titled “Running the Connector”)