Workflow event listener / ConductorQueue in rabbit

[adefretin-adista]

Hi all,

I would like receive a msg in a rabbitmq when a worflow status has changed (failed/completed/finalize).
I found that workflow-event-listener / conductorqueue should be the job but i never received msg on my rabbit when my workflow status changed.

My conductor conf:

# Default event queue type to listen on for wait task
conductor.default-event-queue.type=amqp

conductor.workflow-status-listener.type=queue_publisher
#Queue for successful completion of a workflow
conductor.workflow-status-listener.queue-publisher.successQueue=_callbackSuccessQueue
#Queue for failed workflows
conductor.workflow-status-listener.queue-publisher.failureQueue=_callbackFailureQueue
#Queue for terminal state workflows (success or failed)
conductor.workflow-status-listener.queue-publisher.finalizeQueue=_callbackFinalizeQueue

#Following properties set for using AMQP events and tasks with conductor:
#(To enable support of AMQP queues)
conductor.event-queues.amqp.enabled=true

# Here are the settings with default values
conductor.event-queues.amqp.hosts=rabbitmq:5672
conductor.event-queues.amqp.username=guest
conductor.event-queues.amqp.password=guest

#conductor.event-queues.amqp.virtualHost=/
#conductor.event-queues.amqp.port=5672
#conductor.event-queues.amqp.useNio=false
#conductor.event-queues.amqp.batchSize=1
#conductor.event-queues.amqp.pollTimeDuration=100ms
#conductor.event-queues.amqp.queueType=classic
#conductor.event-queues.amqp.sequentialMsgProcessing=true
#conductor.event-queues.amqp.connectionTimeoutInMilliSecs=180000
#conductor.event-queues.amqp.networkRecoveryIntervalInMilliSecs=5000
#conductor.event-queues.amqp.requestHeartbeatTimeoutInSecs=30
#conductor.event-queues.amqp.handshakeTimeoutInMilliSecs=180000
#conductor.event-queues.amqp.maxChannelCount=5000
#conductor.event-queues.amqp.limit=50
#conductor.event-queues.amqp.duration=1000
#conductor.event-queues.amqp.retryType=REGULARINTERVALS

#conductor.event-queues.amqp.useExchange=true( exchange or queue)
conductor.event-queues.amqp.listenerQueuePrefix=myqueue
# Use durable queue ?
#conductor.event-queues.amqp.durable=false

My wf conf:

...
  "workflowStatusListenerEnabled": true,
...

Conductor logs:

conductor-server           | 243632 [system-task-worker-3] INFO  com.netflix.conductor.contribs.queue.amqp.config.AMQPEventQueueProvider [] - Retrieve queue with URI test-ade-conductor
conductor-server           | 243633 [system-task-worker-3] INFO  com.netflix.conductor.contribs.queue.amqp.AMQPConnection [] - Borrowed the channel object from the channel pool for the connection type [PUBLISHER]
conductor-server           | 243634 [system-task-worker-3] INFO  com.netflix.conductor.contribs.queue.amqp.AMQPConnection [] - Returned the borrowed channel object to the pool for the connection type [PUBLISHER]
conductor-server           | 243634 [system-task-worker-3] INFO  com.netflix.conductor.contribs.queue.amqp.AMQPConnection [] - Borrowed the channel object from the channel pool for the connection type [PUBLISHER]
conductor-server           | 243634 [system-task-worker-3] INFO  com.netflix.conductor.contribs.queue.amqp.AMQPObservableQueue [] - Published message to : {"event_produced":"amqp_queue:test-ade-conductor","asyncComplete":false,"sink":"amqp_queue:test-ade-conductor","workflowType":"test","correlationId":null,"taskToDomain":{},"workflowVersion":2,"workflowInstanceId":"913a2a37-95ef-4176-b8f4-865692ef6bac","param1":"value1"}
conductor-server           | 243634 [system-task-worker-3] INFO  com.netflix.conductor.contribs.queue.amqp.AMQPConnection [] - Returned the borrowed channel object to the pool for the connection type [PUBLISHER]
conductor-server           | 243669 [system-task-worker-3] INFO  com.netflix.conductor.contribs.listener.conductorqueue.ConductorQueueStatusPublisher [] - Publishing callback of workflow 913a2a37-95ef-4176-b8f4-865692ef6bac on completion
conductor-server           | 243672 [system-task-worker-3] INFO  com.netflix.conductor.contribs.listener.conductorqueue.ConductorQueueStatusPublisher [] - Publishing callback of workflow 913a2a37-95ef-4176-b8f4-865692ef6bac on finalization

Connection with rabbit is ok cause i can use an EventTask to publish a msg on it "amqp_queue:test-ade-conductor"

Here are my autocreated rabbit exchanges:

myqueueCOMPLETED 
myqueueFAILED

and queues:

bound_to_myqueueCOMPLETED
bound_to_myqueueFAILED

Thks for your help,

Regards,