Is it possible to use Kafka streams API to reorder out-of-order events?

[Wei Zhou]

Hi,

I wonder if it's possible to re-order a stream of out-of-order events  by leveraging the windowing feature and time semantics of Kafka streams.  By "out of order",  I mean the order of the event time of the records (embedded in the event payload) may be different from the order of event arrival time.    Specially, I  need to able use Kafka API to read the out-of-order event stream and generate an output stream of ordered events and send the new stream back to Kafka, such that  the  order of events  (i.e. order of the message offset ) in the new stream is based on the event time in the message payload, not the order  the events are received in the original topic.  

 

Is it possible to do this using Kafka Streams API?  For example, by assigning event-time to the stream records using Timestamp Extractor, and then self-joining 2 copies of the original streams using a sliding window?  Will this be able to generate an output stream of ordered events?  If this won't work,  is there any other way to make this work using Kafka Streams?

Thanks in advance!

Wei

[Matthias J. Sax]

You can do this, but you need to implement it by your own.

You would use a custom stateful `Transformer`. Within the transformer
you can buffer up records in your state and if you are sure you got all
data, you can emit the records in any order you want.

Using DSL would not be a good approach because there, records are always
processed in offset-order and it's not possible to change this at DSL
level. A custom TimestampExtractor won't help either.


-Matthias

> --
> You received this message because you are subscribed to the Google
> Groups "Confluent Platform" group.
> To unsubscribe from this group and stop receiving emails from it, send
> an email to confluent-platf...@googlegroups.com
> <mailto:confluent-platf...@googlegroups.com>.
> To post to this group, send email to confluent...@googlegroups.com
> <mailto:confluent...@googlegroups.com>.
> To view this discussion on the web visit
> https://groups.google.com/d/msgid/confluent-platform/39318c73-8db7-471f-9b9b-7c113f70e381%40googlegroups.com
> <https://groups.google.com/d/msgid/confluent-platform/39318c73-8db7-471f-9b9b-7c113f70e381%40googlegroups.com?utm_medium=email&utm_source=footer>.
> For more options, visit https://groups.google.com/d/optout.

[Wei Zhou]

Thanks a lot  Matthias!

Wei

[Alper Kanat]

Hi Matthias,

I've a microservice which receives batched events and writes them into a topic one by one using Spring Kafka. Before writing messages to the topic, I calculate the timestamp of each message using their attributes and assign the timestamp to the corresponding ProducerRecord. I also key these messages with some ID within the message. Then another microservice consumes this topic, rekey the messages using another attribute and the KStreams app is configured with a special TimestampExtractor class which extracts the timestamp.

I wonder if the messages are ordered (using the timestamps I assigned earlier) during the repartioning process?

> an email to confluent-platform+unsub...@googlegroups.com
> <mailto:confluent-platform+unsub...@googlegroups.com>.

[Matthias J. Sax]

No.

If data is repartitioned, there is no guarantee that all records in a
partition will be ordered by timestamp. The reason is, that multiple
upstream producers write into the repartition topic, and thus data in
appended interleaved. Hence, even if the data that each producer write
is ordered by timestamp, this property is lost on write into the
repartition topic.


-Matthias

> > an email to confluent-platf...@googlegroups.com
> <javascript:>
> > <mailto:confluent-platf...@googlegroups.com
> <javascript:>>.

> > To post to this group, send email to confluent...@googlegroups.com

> <javascript:>
> > <mailto:confluent...@googlegroups.com <javascript:>>.

> > To view this discussion on the web visit
> >
> https://groups.google.com/d/msgid/confluent-platform/39318c73-8db7-471f-9b9b-7c113f70e381%40googlegroups.com
> <https://groups.google.com/d/msgid/confluent-platform/39318c73-8db7-471f-9b9b-7c113f70e381%40googlegroups.com>
>
> >

> <https://groups.google.com/d/msgid/confluent-platform/39318c73-8db7-471f-9b9b-7c113f70e381%40googlegroups.com?utm_medium=email&utm_source=footer

> <https://groups.google.com/d/msgid/confluent-platform/39318c73-8db7-471f-9b9b-7c113f70e381%40googlegroups.com?utm_medium=email&utm_source=footer>>.
>
> > For more options, visit https://groups.google.com/d/optout

> <https://groups.google.com/d/optout>.

>
> --
> You received this message because you are subscribed to the Google
> Groups "Confluent Platform" group.
> To unsubscribe from this group and stop receiving emails from it, send

> an email to confluent-platf...@googlegroups.com
> <mailto:confluent-platf...@googlegroups.com>.

> To post to this group, send email to confluent...@googlegroups.com
> <mailto:confluent...@googlegroups.com>.
> To view this discussion on the web visit

> https://groups.google.com/d/msgid/confluent-platform/0c7de2a3-6e7c-4e50-a3fe-29a2377cbcc7%40googlegroups.com
> <https://groups.google.com/d/msgid/confluent-platform/0c7de2a3-6e7c-4e50-a3fe-29a2377cbcc7%40googlegroups.com?utm_medium=email&utm_source=footer>.