conductor-oss / conductor-oss/conductor

Unable to cancel messages when using AMQP queues.

Open
#779 1 comment 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
32.2k
Forks
1k
Avg merge
2d 3h
Merged PRs (30d)
37

Description

### Describe the bug

When the `Event.cancel()` [method](https://github.com/conductor-oss/conductor/blob/main/core/src/main/java/com/netflix/conductor/core/execution/tasks/Event.java#L116-L121) is invoked for an AMQP queue, the relevant `ack` [method](https://github.com/conductor-oss/conductor/blob/main/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java#L190-L201) is called. This code tries to [acknowledge the message](https://github.com/conductor-oss/conductor/blob/main/amqp/src/main/java/com/netflix/conductor/contribs/queue/amqp/AMQPObservableQueue.java#L206-L214) using the `receipt` field from the `Message` instance. When this occurs, the `ack` execution fails because the `Event.cancel()` code has inserted the trace ID (a UUID value) as the value for `receipt`, and AMQP expects the `receipt` field to be convertable to a numeric value.

As a result, we see the following stack trace repeatedly:

```
java.lang.NumberFormatException: For input string: "2307ffa1-7a2d-42c6-a975-5f2773902216"
at java.base/java.lang.NumberFormatException.forInputString(NumberFormatException.java:67) ~[?:?]
at java.base/java.lang.Long.parseLong(Long.java:711) ~[?:?]
at java.base/java.lang.Long.parseLong(Long.java:836) ~[?:?]
at com.netflix.conductor.contribs.queue.amqp.AMQPObservableQueue.ackMsg(AMQPObservableQueue.java:211) ~[conductor-amqp.jar!/:?]
at com.netflix.conductor.contribs.queue.amqp.AMQPObservableQueue.ack(AMQPObservableQueue.java:194) ~[conductor-amqp.jar!/:?]
at com.netflix.conductor.core.execution.tasks.Event.cancel(Event.java:120) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.execution.WorkflowExecutorOps.cancelNonTerminalTasks(WorkflowExecutorOps.java:1225) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.execution.WorkflowExecutorOps.cancelNonTerminalTasks(WorkflowExecutorOps.java:1202) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.execution.WorkflowExecutorOps.terminateWorkflow(WorkflowExecutorOps.java:724) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.execution.WorkflowExecutorOps.terminate(WorkflowExecutorOps.java:1644) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.execution.WorkflowExecutorOps.decide(WorkflowExecutorOps.java:1133) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.execution.WorkflowExecutorOps.decide(WorkflowExecutorOps.java:1061) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.reconciliation.WorkflowSweeper.sweep(WorkflowSweeper.java:94) ~[conductor-core.jar!/:?]
at com.netflix.conductor.core.reconciliation.WorkflowSweeper.sweepAsync(WorkflowSweeper.java:75) ~[conductor-core.jar!/:?]
at jdk.internal.reflect.GeneratedMethodAccessor181.invoke(Unknown Source) ~[?:?]
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
at java.base/java.lang.reflect.Method.invoke(Method.java:569) ~[?:?]
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:355) ~[spring-aop-6.1.14.jar!/:6.1.14]
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196) ~[spring-aop-6.1.14.jar!/:6.1.14]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-6.1.14.jar!/:6.1.14]
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:768) ~[spring-aop-6.1.14.jar!/:6.1.14]
at org.springframework.aop.interceptor.AsyncExecutionInterceptor.lambda$invoke$0(AsyncExecutionInterceptor.java:113) ~[spring-aop-6.1.14.jar!/:6.1.14]
at org.springframework.util.concurrent.FutureUtils.lambda$toSupplier$0(FutureUtils.java:74) ~[spring-core-6.1.14.jar!/:6.1.14]
at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
at java.base/java.lang.Thread.run(Thread.java:840) [?:?]
```

The AMQP integration needs to ensure that cancellation messages contain a parseable delivery tag to ensure that the `ack` invocation works as expected.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.