Kogito Quarkus Knative integration: throwing and catching intermediate events

[Constanza]

Hi, 

I created a process that uses messaging start and end events for communication, integrated it with Knative Eventing and tested the functionality - which works as expected. 

Now I'm trying to incorporate the functionality of throwing and catching intermediate message events. When doing so, the first (throwing) works as expected, but when trying to catch an event I get the following error:

2021-07-06 13:43:32,134 DEBUG [org.kie.kog.eve.imp.CloudEventConsumer] (executor-thread-1) Received message without reference id, starting new process instance with 

trigger 'processedcodeset'

2021-07-06 13:43:32,142 ERROR [io.qua.mut.run.MutinyInfrastructure] (executor-thread-1) Mutiny had to drop the following exception: java.lang.IllegalArgumentException: There is no start node that matches the trigger processedcodeset

        at org.jbpm.ruleflow.instance.RuleFlowProcessInstance.internalStart(RuleFlowProcessInstance.java:39)

        at org.jbpm.process.instance.impl.ProcessInstanceImpl.start(ProcessInstanceImpl.java:232)

        at org.jbpm.workflow.instance.impl.WorkflowProcessInstanceImpl.start(WorkflowProcessInstanceImpl.java:473)

        at org.jbpm.process.instance.LightProcessRuntime.startProcessInstance(LightProcessRuntime.java:170)

        at org.jbpm.process.instance.LightProcessRuntime.startProcessInstance(LightProcessRuntime.java:157)

        at org.kie.kogito.process.impl.AbstractProcessInstance.start(AbstractProcessInstance.java:212)

        at org.kie.kogito.event.impl.CloudEventConsumer.lambda$consume$0(CloudEventConsumer.java:84)

        at org.kie.kogito.services.uow.UnitOfWorkExecutor.executeInUnitOfWork(UnitOfWorkExecutor.java:33)

        at org.kie.kogito.event.impl.CloudEventConsumer.consume(CloudEventConsumer.java:62)

        at org.kie.kogito.services.event.impl.AbstractMessageConsumer.consume(AbstractMessageConsumer.java:87)

        at io.smallrye.mutiny.subscription.Subscribers$CallbackBasedSubscriber.onItem(Subscribers.java:67)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.subscription.SafeSubscriber.onNext(SafeSubscriber.java:98)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.operators.multi.multicast.MultiPublishOp$PublishSubscriber.drain(MultiPublishOp.java:475)

        at io.smallrye.mutiny.operators.multi.multicast.MultiPublishOp$PublishSubscriber.onItem(MultiPublishOp.java:182)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.operators.multi.MultiMapOp$MapProcessor.onItem(MultiMapOp.java:50)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.operators.multi.MultiOnItemInvoke$MultiOnItemInvokeProcessor.onItem(MultiOnItemInvoke.java:40)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.subscription.SafeSubscriber.onNext(SafeSubscriber.java:98)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.subscription.SafeSubscriber.onNext(SafeSubscriber.java:98)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.subscription.SafeSubscriber.onNext(SafeSubscriber.java:98)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.operators.multi.MultiOperatorProcessor.onItem(MultiOperatorProcessor.java:66)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.lambda$onNext$1(ContextPropagationMultiInterceptor.java:89)    

        at io.smallrye.context.SmallRyeThreadContext.lambda$withContext$1(SmallRyeThreadContext.java:530)

        at io.smallrye.mutiny.context.ContextPropagationMultiInterceptor$ContextPropagationSubscriber.onNext(ContextPropagationMultiInterceptor.java:89)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)

        at io.smallrye.mutiny.helpers.HalfSerializer.onNext(HalfSerializer.java:31)

        at io.smallrye.mutiny.helpers.StrictMultiSubscriber.onItem(StrictMultiSubscriber.java:83)

        at io.smallrye.mutiny.operators.multi.builders.BufferItemMultiEmitter.drain(BufferItemMultiEmitter.java:109)

        at io.smallrye.mutiny.operators.multi.builders.BufferItemMultiEmitter.emit(BufferItemMultiEmitter.java:34)

        at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.onItem(SerializedMultiEmitter.java:49)

        at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.emit(SerializedMultiEmitter.java:136)

        at io.smallrye.reactive.messaging.extension.ThrowingEmitter.emit(ThrowingEmitter.java:63)

        at io.smallrye.reactive.messaging.extension.AbstractEmitter.emit(AbstractEmitter.java:146)

        at io.smallrye.reactive.messaging.extension.EmitterImpl.send(EmitterImpl.java:29)

        at org.kie.kogito.app.CloudEventListenerResource.cloudEventListener(CloudEventListenerResource.java:49)

        at org.kie.kogito.app.CloudEventListenerResource_ClientProxy.cloudEventListener(CloudEventListenerResource_ClientProxy.zig:186)

        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

        at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)

        at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)

        at java.lang.reflect.Method.invoke(Unknown Source)

        at org.jboss.resteasy.core.MethodInjectorImpl.invoke(MethodInjectorImpl.java:170)

        at org.jboss.resteasy.core.MethodInjectorImpl.invoke(MethodInjectorImpl.java:130)

        at org.jboss.resteasy.core.ResourceMethodInvoker.internalInvokeOnTarget(ResourceMethodInvoker.java:643)

        at org.jboss.resteasy.core.ResourceMethodInvoker.invokeOnTargetAfterFilter(ResourceMethodInvoker.java:507)

        at org.jboss.resteasy.core.ResourceMethodInvoker.lambda$invokeOnTarget$2(ResourceMethodInvoker.java:457)

        at org.jboss.resteasy.core.interception.jaxrs.PreMatchContainerRequestContext.filter(PreMatchContainerRequestContext.java:364)

        at org.jboss.resteasy.core.ResourceMethodInvoker.invokeOnTarget(ResourceMethodInvoker.java:459)

        at org.jboss.resteasy.core.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:419)

        at org.jboss.resteasy.core.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:393)

        at org.jboss.resteasy.core.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:68)

        at org.jboss.resteasy.core.SynchronousDispatcher.invoke(SynchronousDispatcher.java:492)

        at org.jboss.resteasy.core.SynchronousDispatcher.lambda$invoke$4(SynchronousDispatcher.java:261)

        at org.jboss.resteasy.core.SynchronousDispatcher.lambda$preprocess$0(SynchronousDispatcher.java:161)

        at org.jboss.resteasy.core.interception.jaxrs.PreMatchContainerRequestContext.filter(PreMatchContainerRequestContext.java:364)

        at org.jboss.resteasy.core.SynchronousDispatcher.preprocess(SynchronousDispatcher.java:164)

        at org.jboss.resteasy.core.SynchronousDispatcher.invoke(SynchronousDispatcher.java:247)

        at org.jboss.resteasy.plugins.server.servlet.ServletContainerDispatcher.service(ServletContainerDispatcher.java:249)

        at io.quarkus.resteasy.runtime.ResteasyFilter.doFilter(ResteasyFilter.java:35)

        at io.undertow.servlet.core.ManagedFilter.doFilter(ManagedFilter.java:61)

        at io.undertow.servlet.handlers.FilterHandler$FilterChainImpl.doFilter(FilterHandler.java:131)

        at io.undertow.servlet.handlers.FilterHandler.handleRequest(FilterHandler.java:84)

        at io.undertow.servlet.handlers.security.ServletSecurityRoleHandler.handleRequest(ServletSecurityRoleHandler.java:63)

        at io.undertow.servlet.handlers.ServletChain$1.handleRequest(ServletChain.java:68)

        at io.undertow.servlet.handlers.ServletDispatchingHandler.handleRequest(ServletDispatchingHandler.java:36)

        at io.undertow.servlet.handlers.RedirectDirHandler.handleRequest(RedirectDirHandler.java:67)

        at io.undertow.servlet.handlers.security.SSLInformationAssociationHandler.handleRequest(SSLInformationAssociationHandler.java:133)

        at io.undertow.servlet.handlers.security.ServletAuthenticationCallHandler.handleRequest(ServletAuthenticationCallHandler.java:57)

        at io.undertow.server.handlers.PredicateHandler.handleRequest(PredicateHandler.java:43)

        at io.undertow.security.handlers.AbstractConfidentialityHandler.handleRequest(AbstractConfidentialityHandler.java:46)

        at io.undertow.servlet.handlers.security.ServletConfidentialityConstraintHandler.handleRequest(ServletConfidentialityConstraintHandler.java:65)

        at io.undertow.security.handlers.AuthenticationMechanismsHandler.handleRequest(AuthenticationMechanismsHandler.java:60)

        at io.undertow.servlet.handlers.security.CachedAuthenticatedSessionHandler.handleRequest(CachedAuthenticatedSessionHandler.java:77)

        at io.undertow.security.handlers.NotificationReceiverHandler.handleRequest(NotificationReceiverHandler.java:50)

        at io.undertow.security.handlers.AbstractSecurityContextAssociationHandler.handleRequest(AbstractSecurityContextAssociationHandler.java:43)

        at io.undertow.server.handlers.PredicateHandler.handleRequest(PredicateHandler.java:43)

        at io.undertow.server.handlers.PredicateHandler.handleRequest(PredicateHandler.java:43)

        at io.undertow.servlet.handlers.ServletInitialHandler.handleFirstRequest(ServletInitialHandler.java:247)

        at io.undertow.servlet.handlers.ServletInitialHandler.access$100(ServletInitialHandler.java:56)

        at io.undertow.servlet.handlers.ServletInitialHandler$2.call(ServletInitialHandler.java:111)

        at io.undertow.servlet.handlers.ServletInitialHandler$2.call(ServletInitialHandler.java:108)

        at io.undertow.servlet.core.ServletRequestContextThreadSetupAction$1.call(ServletRequestContextThreadSetupAction.java:48)

        at io.undertow.servlet.core.ContextClassLoaderSetupAction$1.call(ContextClassLoaderSetupAction.java:43)

        at io.quarkus.undertow.runtime.UndertowDeploymentRecorder$9$1.call(UndertowDeploymentRecorder.java:572)

        at io.undertow.servlet.handlers.ServletInitialHandler.dispatchRequest(ServletInitialHandler.java:227)

        at io.undertow.servlet.handlers.ServletInitialHandler.handleRequest(ServletInitialHandler.java:152)

        at io.quarkus.undertow.runtime.UndertowDeploymentRecorder$1.handleRequest(UndertowDeploymentRecorder.java:117)

        at io.undertow.server.Connectors.executeRootHandler(Connectors.java:290)

        at io.undertow.server.DefaultExchangeHandler.handle(DefaultExchangeHandler.java:18)

        at io.quarkus.undertow.runtime.UndertowDeploymentRecorder$5$1.run(UndertowDeploymentRecorder.java:398)

        at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)

        at java.util.concurrent.FutureTask.run(Unknown Source)

        at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2415)

        at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1452)

        at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:29)

        at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:29)

        at java.lang.Thread.run(Unknown Source)

        at org.jboss.threads.JBossThread.run(JBossThread.java:501)

I see there's a log mentioning the message was received without a reference id, but I did include the process instance id as a cloud event header, as well as the same value for the ce-kogitoreferenceid header. 

What is the right way to send the cloud event including a reference id so that it does not look for a start node that matches the trigger, but matches the reference id to the running process and uses it to send the triggering signal for that intermediate event?

Thanks,

C

[Ricardo Zanini]

Hi Constanza!

Many thanks for trying Kogito with Knative. I think the header for the Kogito reference id extension is wrong. Try "kogitoprocrefid"

We changed these names a while back to align with the CloudEvents spec. Unfortunately, the documentation hasn't been updated yet. I'll fix this tomorrow.

Thanks!

--

Zanini

--
You received this message because you are subscribed to the Google Groups "Kogito development mailing list" group.
To unsubscribe from this group and stop receiving emails from it, send an email to kogito-developm...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/kogito-development/873b47e3-53c8-4a1b-aff1-3c2b3de45354n%40googlegroups.com.

[Constanza]

Thank you, Ricardo!

I tried adding the ce-kogitoprocrefid instead of the ce-kogitoreferenceid I was using previously, you can check below, but I still get the same output.

2021-07-06 16:39:45,567 DEBUG [CloudEventListenerResource] (executor-thread-1) CloudEvent received:

? ?cloudevents.Event

Context Attributes,

        specversion: 1.0

        type: processedcodeset

        source: dev.postman.CF

        id: 048b8b03-8ec0-4138-b470-8fd0fe09ee45

Extensions,     kogitoprocrefid: 12345

        kogitoprocinstanceid: 2f3d89a0-2cd8-4109-9c7b-3371ae206405

        kogitoprocist: 1

        kogitoprocid: claims

Data,

...

2021-07-06 16:39:45,579 DEBUG [org.kie.kog.eve.imp.CloudEventConsumer] (executor-thread-1) Received message without reference id, starting new process instance with 

trigger 'processedcodeset'

2021-07-06 16:39:45,580 ERROR [io.qua.mut.run.MutinyInfrastructure] (executor-thread-1) Mutiny had to drop the following exception: java.lang.IllegalArgumentException: There is no start node that matches the trigger processedcodeset

        at org.jbpm.ruleflow.instance.RuleFlowProcessInstance.internalStart(RuleFlowProcessInstance.java:39)

...

Am I missing something?

Thanks,

Constanza

[Ricardo Zanini]

Hi Constanza,

It might be a bug, do you mind sharing a reproducer so I can take a look?

--

Zanini

To view this discussion on the web visit https://groups.google.com/d/msgid/kogito-development/01274a44-ed54-41b2-8045-1d06c1a47e64n%40googlegroups.com.

[Constanza]

Hi Ricardo, 

Sure! Here you go: https://github.com/constanzafc/sample-workflow

Thanks,

C

[Constanza]

Please ignore ^ found the bug, fixed it and confirmed it works. Thanks for the help!

[Ricardo Zanini]

Glad that works! What did you do?

:)

To view this discussion on the web visit https://groups.google.com/d/msgid/kogito-development/0610e0dd-dc53-48f7-bb23-ab9a6714afecn%40googlegroups.com.

--

--

Zanini