multiple kogito_incoming_stream

[Mark Jayson Gonzaga]

Hi Team,

I would like to ask for expert advice regarding the problem we are facing.

However, I observed that in the above image I have to define another  kogito_incoming_stream  which I already did for the  intermediate  message after the Validate Contract Form User task.

mp.messaging.incoming.kogito_incoming_stream.connector=smallrye-amqp
mp.messaging.incoming.kogito_incoming_stream.address=servicecontractresponses

mp.messaging.outgoing.kogito_outgoing_stream.connector=smallrye-amqp
mp.messaging.outgoing.kogito_outgoing_stream.address=servicecontractapplications

## metadata
mp.messaging.outgoing.kogito-processinstances-events.connector=smallrye-amqp
mp.messaging.outgoing.kogito-processinstances-events.events=kogito-processinstances-events

My question is how can I add another kogito_incoming_stream for my 2nd intermediate message after the PullOut/Internment User task?

Thanks for the help in advance.

Regards

[Francisco Javier Tirado Sarti]

Hi Mark,

Take a look to the multi addon example https://github.com/kiegroup/kogito-examples/blob/stable/process-kafka-multi-quarkus/src/main/resources/application.properties .

It is for publishing to multiple topics, but the idea for subscription is the same. You use the trigger name as discriminator in the properties, rather than kogito-incoming_stream

The dependency for the addon is

    <dependency>
    <groupId>org.kie.kogito</groupId>
    <artifactId>kogito-addons-quarkus-cloudevents-multi</artifactId>
    </dependency>

[Mark Jayson Gonzaga]

Hi Francisco,

When I updated my dependency I got the below error.

2021-07-14 02:14:11,509 ERROR [io.qua.run.Application] (Quarkus Main Thread) Failed to start application (with profile dev): javax.enterprise.inject.spi.DefinitionException: SRMSG00019: Unable to connect an emitter with the channel `kogito-processinstances-events`
at io.smallrye.reactive.messaging.extension.ChannelProducer.getEmitter(ChannelProducer.java:179)
at io.smallrye.reactive.messaging.extension.ChannelProducer.produceEmitter(ChannelProducer.java:124)
at io.smallrye.reactive.messaging.extension.ChannelProducer_Subclass.produceEmitter$$superforward1(ChannelProducer_Subclass.zig:565)
at io.smallrye.reactive.messaging.extension.ChannelProducer_Subclass$$function$$10.apply(ChannelProducer_Subclass$$function$$10.zig:33)

Below is my pom.xml

<dependencies>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-arc</artifactId>
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-grpc</artifactId>
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-smallrye-openapi</artifactId>
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-smallrye-health</artifactId>
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-smallrye-reactive-messaging-amqp</artifactId> <!--amqp-->
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-mailer</artifactId>
  </dependency>
  <!--KOGITO RELATED DEPENDENCIES-->
  <dependency>
    <groupId>org.kie.kogito</groupId>
    <artifactId>kogito-quarkus</artifactId>
  </dependency>

  <dependency>
    <groupId>org.kie.kogito</groupId>
    <artifactId>kogito-addons-quarkus-cloudevents-multi</artifactId>
  </dependency>
  <dependency>
    <groupId>org.kie.kogito</groupId>

    <artifactId>kogito-addons-quarkus-events-smallrye</artifactId>
  </dependency>
  <dependency>
    <groupId>org.kie.kogito</groupId>
    <artifactId>kogito-addons-quarkus-process-management</artifactId>
  </dependency>
  <dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.20</version>
    <scope>provided</scope>
  </dependency>
</dependencies>

And my property file

mp.messaging.incoming.kogito_incoming_stream.connector=smallrye-amqp
mp.messaging.incoming.kogito_incoming_stream.address=servicecontractresponses

mp.messaging.outgoing.kogito_outgoing_stream.connector=smallrye-amqp
mp.messaging.outgoing.kogito_outgoing_stream.address=servicecontractapplications

## metadata
mp.messaging.outgoing.kogito-processinstances-events.connector=smallrye-amqp
mp.messaging.outgoing.kogito-processinstances-events.events=kogito-processinstances-events

mp.messaging.outgoing.kogito-usertaskinstances-events.connector=smallrye-amqp
mp.messaging.outgoing.kogito-usertaskinstances-events.events=kogito-usertaskinstances-events

mp.messaging.outgoing.kogito-variables-events.connector=smallrye-amqp
mp.messaging.outgoing.kogito-variables-events.address=kogito-variables-events

But when I removed this

<!--    <dependency>-->
<!--      <groupId>org.kie.kogito</groupId>-->
<!--      <artifactId>kogito-addons-quarkus-events-smallrye</artifactId>-->
<!--    </dependency>-->

Looks like the error went away.

Maybe I don't really understand what that addons does

--
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/0d8be3ae-1785-4559-8ad6-7fa6465b5b17n%40googlegroups.com.

[Mark Jayson Gonzaga]

Btw,

What would be the common issue for this kind of message

Not sure where it got the efbacca8-40ad-4e05-9a37-ab32c9319fa4. but below are the actual id's I entered Upon invoking the Usertask which eventually send message

And it gave me 200.

But on my last service I got the "efbacca8-40ad-4e05-9a37-ab32c9319fa4"

my bpmn

Thanks for the response.

[Francisco Javier Tirado Sarti]

Hi, 

First the feedback about your configuration, which you need to modify. 

What multi addon does is rather than using just one incoming and outgoing channel (which are named kogito_incoming_stream and kogito_output_stream) for processing all events related with messages nodes defined in your bpmn, it tells Kogito to use as many channels as message names you have defined in your bpmn.

Therefore, once you update your dependency, you need to change your property file accordingly. In particular you need to add configuration entries with the name of every message (incoming or outcoming) that you are using in your code (currently defined kogito_incoming_stream and kogito_outgoing_stream configuration entries are ignored by multi addon, you can leave it there in case you switch back to "single" addon)

For example if your if your first incoming message "Approved" has a name "approved" you need to use

mp.messaging.incoming.approved.connector=smallrye-amqp
mp.messaging.incoming.aproved.address=servicecontractresponses

You need to do that for every message name, so

 if your second incoming message "Rejected" has a name "rejected" you need to specify

mp.messaging.incoming.rejected.connector=smallrye-amqp
mp.messaging.incoming.rejected.address=<your second address>

Note that if all your address are the same, you do not need the multi addon at all.

Now, regarding the messsage you are seeing, Im not 100% sure (Im not familiar with AMQP over smallryy) but I would say the problem is that you have not defined the address for channel kogito-processinstances-event (note that you are using "events" rather than "address", as you are doing for other channels). Anyway, it seems you are not interested on receiving notification about process instance, task or variable changes, therefore you can comment "kogito-addons-quarkus-events-smallrye" dependency as you did.

Hope this clarifies.

[Mark Jayson Gonzaga]

Hi Francisco,

Thanks for the wonderful explanation. That really helps adding more insights about kogit.

Btw, I'm trying to create an orchestration based microservice. Is there a better way to do it? or Is my approach ok?

To view this discussion on the web visit https://groups.google.com/d/msgid/kogito-development/def1e5ce-881b-4d77-a90f-981e3452bbb5n%40googlegroups.com.

[Mark Jayson Gonzaga]

Hi Francisco,

I noticed something.

When I trigger the API for the highlighted User task

It printed 

2021-07-15 00:34:11,218 WARN  [org.kie.kog.eve.imp.CloudEventConsumer] (pool-1-thread-1) Consumer for CloudEvent type 'MfhsMessageDataEvent_6', trigger 'pulloutinternmentrequestsapproved': ignoring message with type 'servicecontractapproved',  source '/process/service_contract'
2021-07-15 00:34:11,222 WARN  [org.kie.kog.eve.imp.CloudEventConsumer] (pool-1-thread-1) Consumer for CloudEvent type 'MfhsMessageDataEvent_11', trigger 'servicecontractrejected': ignoring message with type 'servicecontractapproved',  source '/process/service_contract'
2021-07-15 00:34:11,224 WARN  [org.kie.kog.eve.imp.CloudEventConsumer] (pool-1-thread-1) Consumer for CloudEvent type 'MfhsMessageDataEvent_5', trigger 'pulloutinternmentrequestsrejected': ignoring message with type 'servicecontractapproved',  source '/process/service_contract'

Now the  pulloutinternmentrequestsapproved and  pulloutinternmentrequestsrejected are after the PullOutInternment User Task.

Since I only invoked the User task prior to that PullOutInternment my assumption is these 2 events/messages should not be triggered.

Is that an expected behaviour?

[Mark Jayson Gonzaga]

I added a reproducer using kafka 

Is Multiple Gateway Event Supported? · Issue #794 · kiegroup/kogito-examples (github.com)

[Francisco Javier Tirado Sarti]

HI Mark,

You can quietly ignore these logs. They should not affect the behaviour of the process and they'll be gone once Kogito-5252 is merged. Before that JIRA, multi addon use a single queue for all registrations, hence those messages (which just indicates that the event being consumed should not be considered for those registrations). With Kogito-5252, the approach used to handle events internally is different and the event will be directly mapped to the expected registration, so these logs (which I agree are confusing, which is one of the reasons for Kogito-5252 ;)) will not longer appear

[Mark Jayson Gonzaga]

I see will that be included in the next release?

To view this discussion on the web visit https://groups.google.com/d/msgid/kogito-development/04835661-ed28-4853-a516-bd1c40b50c5fn%40googlegroups.com.