How to configure amqp queue connection?

[Jurgen1980]

I'm trying to send messages to a RabbitMQ queue.
But I keep getting the same error over and over again.

Error:

Error loading queue for name:amqp_queue:conductor, sink:amqp_queue:conductor, error: Unknown queue type amqp_queue

Flow:

{
  "createTime": 1658226140596,
  "name": "Workflow01",
  "description": "",
  "version": 1,
  "tasks": [
    {
      "name": "Task01_ampq",
      "taskReferenceName": "Task01_ampq_ref",
      "inputParameters": {},
      "type": "EVENT",
      "decisionCases": {},
      "defaultCase": [],
      "forkTasks": [],
      "startDelay": 0,
      "joinOn": [],
      "sink": "amqp_queue:conductor",
      "optional": false,
      "defaultExclusiveJoinTask": [],
      "asyncComplete": false,
      "loopOver": []
    }
  ],
  "inputParameters": [],
  "outputParameters": {},
  "schemaVersion": 2,
  "restartable": true,
  "workflowStatusListenerEnabled": true,
  "ownerEmail": "[email protected]",
  "timeoutPolicy": "ALERT_ONLY",
  "timeoutSeconds": 10,
  "variables": {},
  "inputTemplate": {}
}

What have I done:

added the contrib files

https://github.com/Netflix/conductor/tree/2.31/contribs/src/main/java/com/netflix/conductor/contribs

Configured config file:

conductor.event-queues.amqp.enabled=true
conductor.event-queues.amqp.hosts=<Host IP>
conductor.event-queues.amqp.port=1883
conductor.event-queues.amqp.username=<login>
conductor.event-queues.amqp.password=<password>
conductor.event-queues.amqp.virtualhost=conductor
conductor.event-queues.amqp.connectionTimeout=60000
conductor.event-queues.amqp.useExchange=true
conductor.event-queues.amqp.listenerQueuePrefix=

Verified configuration in swagger:

{
  "conductor.event-queues.amqp.username": "<login>",
  "awt.toolkit": "sun.awt.X11.XToolkit",
  "conductor.elasticsearch.clusterHealthColor": "yellow",
  "java.specification.version": "11",
  "sun.cpu.isalist": "",
  "CONDUCTOR_CONFIG_FILE": "/app/config/config-local.properties",
  "sun.jnu.encoding": "UTF-8",
  "java.class.path": "conductor-server-3.11.0-SNAPSHOT-boot.jar",
  "java.vm.vendor": "Oracle Corporation",
  "sun.arch.data.model": "64",
  "conductor.event-queues.amqp.hosts": "<host IP>",
  "java.vendor.url": "https://openjdk.java.net/",
  "conductor.indexing.enabled": "true",
  "catalina.useNaming": "false",
  "user.timezone": "Etc/UTC",
  "os.name": "Linux",
  "java.vm.specification.version": "11",
  "sun.java.launcher": "SUN_STANDARD",
  "queues.dynomite.threads": "10",
  "sun.boot.library.path": "/usr/local/openjdk-11/lib",
  "sun.java.command": "conductor-server-3.11.0-SNAPSHOT-boot.jar",
  "jdk.debug": "release",
  "conductor.db.type": "memory",
  "sun.cpu.endian": "little",
  "loadSample": "false",
  "user.home": "/root",
  "user.language": "en",
  "java.specification.vendor": "Oracle Corporation",
  "java.version.date": "2022-04-19",
  "java.home": "/usr/local/openjdk-11",
  "version": "3.11.0-SNAPSHOT",
  "conductor.elasticsearch.indexReplicasCount": "0",
  "file.separator": "/",
  "conductor.redis.queuesNonQuorumPort": "22122",
  "java.vm.compressedOopsMode": "Zero based",
  "line.separator": "\n",
  "java.specification.name": "Java Platform API Specification",
  "java.vm.specification.vendor": "Oracle Corporation",
  "conductor.event-queues.amqp.virtualhost": "conductor",
  "FILE_LOG_CHARSET": "UTF-8",
  "java.awt.graphicsenv": "sun.awt.X11GraphicsEnvironment",
  "conductor.redis.workflowNamespacePrefix": "conductor",
  "java.awt.headless": "true",
  "conductor.app.workflowRepairServiceEnabled": "true",
  "java.protocol.handler.pkgs": "org.springframework.boot.loader",
  "sun.management.compiler": "HotSpot 64-Bit Tiered Compilers",
  "java.runtime.version": "11.0.15+10",
  "user.name": "root",
  "buildDate": "2022-07-19T09:42:44.501Z",
  "path.separator": ":",
  "conductor.elasticsearch.url": "http://es:9200",
  "os.version": "5.10.102.1-microsoft-standard-WSL2",
  "java.runtime.name": "OpenJDK Runtime Environment",
  "file.encoding": "UTF-8",
  "spring.beaninfo.ignore": "true",
  "conductor.event-queues.amqp.useExchange": "true",
  "java.vm.name": "OpenJDK 64-Bit Server VM",
  "java.vendor.version": "18.9",
  "java.vendor.url.bug": "https://bugreport.java.com/bugreport/",
  "java.io.tmpdir": "/tmp",
  "conductor.event-queues.amqp.password": "<password>",
  "catalina.home": "/tmp/tomcat.8080.11422540242885503534",
  "java.version": "11.0.15",
  "user.dir": "/app/libs",
  "os.arch": "amd64",
  "conductor.event-queues.amqp.enabled": "true",
  "conductor.grpc-server.enabled": "false",
  "java.vm.specification.name": "Java Virtual Machine Specification",
  "PID": "7",
  "java.awt.printerjob": "sun.print.PSPrinterJob",
  "sun.os.patch.level": "unknown",
  "CONSOLE_LOG_CHARSET": "UTF-8",
  "conductor.event-queues.amqp.listenerQueuePrefix": "",
  "catalina.base": "/tmp/tomcat.8080.11422540242885503534",
  "java.library.path": "/usr/java/packages/lib:/usr/lib64:/lib64:/lib:/usr/lib",
  "java.vm.info": "mixed mode, sharing",
  "java.vendor": "Oracle Corporation",
  "java.vm.version": "11.0.15+10",
  "sun.io.unicode.encoding": "UnicodeLittle",
  "java.class.version": "55.0",
  "conductor.redis.queueNamespacePrefix": "conductor_queues",
  "conductor.redis.hosts": "dyno1:8102:us-east-1c",
  "conductor.event-queues.amqp.connectionTimeout": "60000"
}

Installed version:
(v3.11.0-SNAPSHOT) :::Spring Boot::: (v2.6.7)
Is there something I missed?

[manan164]

Hi @Jurgen1980 , This error comes when there is no AMQPEventQueueProvider initialize. Can you please check do you have in AMQPEventQueueProvider.java your contrib?

[Jurgen1980]

I can confirm I have that file in that location.
But I use Conductor v3.10.0. Don't know if that is compatible?

[manan164]

Quick way to check this is, whether the constructor getting initialized or not via debug point. I am not able to find that file in 3.10.0

[aravindanr]

AMQP provider is moved to the conductor-community repo.