KTable "compare and swap"

[Matt Stern <mj...@gmail.com>]

Hello all,

I am curious if it's possible to use the Streams DSL to perform a "compare
and swap" operation. To give a concrete example:

   - User orders are written to an input topic
   - Our Kafka Streams app reads the order topic and sends each order for
   processing.
   - For a given user, only one order can be sent for processing at a time.
   If a user sends another order while a previous order is still processing,
   we reject it.

I was thinking of using a KTable to store the "state" of each user, where
the key is a user ID and the value is whether an order is out for
processing.

When a new order comes in, we would do a conceptual "compare and swap" on
its value -- if we change the value from false to true, then we send the
order for processing. Otherwise, we "reject" the order.

Looking at it another way, we are using our KTable as a per-user state
machine and we want to respond to particular state changes (e.g. no order
processing -> order processing).

Based on Stack Overflow, it seems like this is possible with the Processor
API (1
<https://stackoverflow.com/questions/49492736/writing-directly-to-a-kafka-state-store>,
2 <https://stackoverflow.com/a/58747230>), but I am curious if it is
possible or recommended to do this with the Streams DSL.

Thanks!
Matt

Re: KTable "compare and swap"

[Matt Stern <mj...@gmail.com>]

I'm wondering if the following works:

   - Make the KTable value a Pair<Event, Event>, where Event is some enum
   like {NO_ORDER, ORDER_RECEIVED, ORDER_PROCESSED}
   - This pair will store the last two events that occurred for a
   particular user.
   - We use the reduce() function to change the Pair value:
      - Suppose for user U, we start with <NO_ORDER, NO_ORDER>
      - When we get an order, we use reduce() to update the KTable value to
      <ORDER_RECEIVED, NO_ORDER>
      - When we get another order, we use reduce() to update the KTable to
      <ORDER_RECEIVED, ORDER_RECEIVED>
      - When an order is processed, we use reduced() to update the KTable
      to <ORDER_PROCESSED, ORDER_RECEIVED>

Now, we can pass the KTable into a filter() operation and send out
processing requests only when the KTable value is <ORDER_RECEIVED,
NO_ORDER> or <ORDER_RECEIVED, ORDER_PROCESSED>.

On Fri, Nov 6, 2020 at 2:06 PM Matt Stern <mj...@gmail.com> wrote:

> Hello all,
>
> I am curious if it's possible to use the Streams DSL to perform a "compare
> and swap" operation. To give a concrete example:
>
>    - User orders are written to an input topic
>    - Our Kafka Streams app reads the order topic and sends each order for
>    processing.
>    - For a given user, only one order can be sent for processing at a
>    time. If a user sends another order while a previous order is still
>    processing, we reject it.
>
> I was thinking of using a KTable to store the "state" of each user, where
> the key is a user ID and the value is whether an order is out for
> processing.
>
> When a new order comes in, we would do a conceptual "compare and swap" on
> its value -- if we change the value from false to true, then we send the
> order for processing. Otherwise, we "reject" the order.
>
> Looking at it another way, we are using our KTable as a per-user state
> machine and we want to respond to particular state changes (e.g. no order
> processing -> order processing).
>
> Based on Stack Overflow, it seems like this is possible with the Processor
> API (1
> <https://stackoverflow.com/questions/49492736/writing-directly-to-a-kafka-state-store>,
> 2 <https://stackoverflow.com/a/58747230>), but I am curious if it is
> possible or recommended to do this with the Streams DSL.
>
> Thanks!
> Matt
>