State machine won't trigger a schedule set after consuming a Kafka event

[bmmptlgc]

Contact Details

bmendonca@letsgetchecked.com

Version

8.x

On which operating system(s) are you experiencing the issue?

Windows

Using which broker(s) did you encounter the issue?

Amazon SQS

What are the steps required to reproduce the issue?

1. Follow the setup instructions on the git repo
2. Run the MT.Saga.Outbox project to start the saga
3. Run the MT.All.In.One.Service to start publishing events
4. Note the behavior below:

I have a Saga state machine that receives events from Kafka and SNS. For that I configure an SQS bus with a Kafka Rider.

I'm trying to configure schedules in my state machine for 2 different scenarios. As a result of receiving an SNS event and as a result of receiving a Kafka event.

It works when the state machine is consuming an SNS event and sets up a schedule (the .Schedule call is commented in the repo code because I want to demonstrate the other scenario below):

During(Created,
    When(ProductSold) // SNS Event
    ...
        // This is Published to http://localhost:4566/000000000000/order-saga
        .Schedule(OrderExpirationSchedule, context => context.Init<OrderExpired>(new { context.Data.OrderId }))
        .TransitionTo(Sold),
...
During(Sold,
    ....
    When(OrderExpired) // This is Consumed
        .Finalize()
);

When I set the scheduler from consuming a Kafka event it doesn't work:

Initially(
    When(OrderCreated) // Kafka Event
    ....
        .Schedule(ProductSaleExpirationSchedule, context =>  context.Init<ProductFailedToSell>(new { context.Data.OrderId }))
        .TransitionTo(Created)
    ...

During(Created,
    When(ProductFailedToSell) // This is NOT Consumed, possibly trying to consume from /order-saga (not sure)
        .Finalize()

What is the expected behavior?

Initially(
    When(OrderCreated) // Kafka Event
    ....
        // This is Published to http://localhost:4566/000000000000/OrderCreated (wish there was a way to specify /order-saga instead
        .Schedule(ProductSaleExpirationSchedule, context =>  context.Init<ProductFailedToSell>(new { context.Data.OrderId }))
        .TransitionTo(Created)
    ...

During(Created,
    When(ProductFailedToSell) // This should trigger and close the saga
        .Finalize()

What actually happened?

Initially(
    When(OrderCreated) // Kafka Event
    ....
        .Schedule(ProductSaleExpirationSchedule, context =>  context.Init<ProductFailedToSell>(new { context.Data.OrderId }))
        .TransitionTo(Created)
    ...

During(Created,
    When(ProductFailedToSell) // This is NOT Consumed, possibly trying to consume from /order-saga (not sure)
        .Finalize()

Related log output, including any exceptions

No response

Link to repository that demonstrates/reproduces the issue

https://github.com/bmmptlgc/MT.Saga.Outbox

[phatboyg]

Have you reviewed the logs? You can't schedule messages on Kafka topic endpoints, so that's probably why it isn't working from Kafka.

[bmmptlgc]

Looking at the code above, we only know OrderCreated is a result of a Kafka event being consumed because of my comment (// Kafka Event). If I change my producer to send it through SNS, then the schedule would work. My point is, looking at the two snippets of code above, for ProductSold and OrderCreated, they "are" both the same in a sense. Why is the schedule tied to the incoming event source and not to the bus endpoint configured for the saga, in this case, the SQS endpoint http://localhost:4566/000000000000/order-saga?

On another note, if I uncomment the following code on my startup file, it works

                        //amazonSqsConfig.ReceiveEndpoint("OrderCreated", e =>
                        //{
                        //    e.ConfigureSaga<OrderState>(context);
                        //});

This is a receive endpoint on SQS, that has nothing to do with Kafka.

It seems that all SNS events result in the scheduler sending messages to http://localhost:4566/000000000000/order-saga, the endpoint configured for the saga (not and endpoint related to ProductSold in the case of my example).
The scheduler for messages that arrive from kafka is creating endpoints that are related to the events and sending messages there,, when it seems to me that it could just send the messages to the endpoint configured for the saga (http://localhost:4566/000000000000/order-saga). Or maybe there could be a config option so we could pass the endpoint we want the scheduler to use when we set up the scheduler.

[phatboyg]

The rider topic endpoint has no clue that you've configured the saga on SQS, and therefore has no way to know that you've configured any sort of message scheduling. It's a known limitation at this point, and I don't have a fix planned.