Delaying all messages by 30 minutes in Kafka Streams

user_709537 0
Viewed 688

We have a use case where we need to write all messages of topic a into topic b, but with a delay of 30 minutes for each message. Why, you ask? Because time is of critical importance for this stream of data, so paying customers get the real-time feed, for freeloaders, we offer the delayed stream.

I guess it would be relatively easy to do in a KafkaConsumer poll() loop, by comparing system time and message time (using an ordered message time like producer time or ingestion time) and then pause()ing the partitions in question and resume()ing them after the appropriate time interval of up to 30 minutes(, all the while continuing to poll() to avoid getting failed over).

As the data, though delayed, still needs to be delivered in a streaming fashion, the delay of the ingestion times of all messages in topic a and b should be as close to 30 minutes as possible.

  1. But is this also easily possible in Kafka Streams, so that we can use its built-in exactly-once guarantees? I wonder if "it's ok to call Thread.sleep() in Kafka Streams also applies to longer sleeps of up to 30 minutes? (Of course we don't want a partition rebalance to occur because Kafka thinks something's wrong with our process)

  2. Assuming we get this to work, is there a way to get proper lag monitoring for this? If we just delay messages, I would think the consumer group lag would always amount to at least 30 minutes worth of messages. So is it possible to have the lag monitor count only unprocessed messages older than 30 minutes?

(2. is of less importance for us than getting 1. to work)

Edit: https://stackoverflow.com/a/59261274/709537 proposes a solution to a somewhat related problem, but that involves state stores and thus looks more complicated than would seem necessary for our simple (?) "delay all messages by x minutes" task.

apache-kafka-streams apache-kafka
1 Answers

Regarding 2., I assume we will have to roll our own lag monitoring for this.

A simple way to do something related - measuring latency instead of the number of lagging messages - would be to periodically and for every partition

  1. get the first unread input message
  2. currentLatency = max(0, ingestionTime(firstUnreadMessage) - 30min)

If we wanted to monitor the number of lagging messages, something a little more involved would need to be done:

  1. read input messages backwards, until there is one with ingestionTime + 30min <= systemTime
  2. the number of those messages would be the lag

However reading messages backwards is not exactly one of Kafka's core competencies... A clever binary search style could be devised to get the exact value. However, no-one really wants to know whether the message lag is 43123 or 40513, what they want to know is the order of magnitude. That will keep the number of seeks down to a handful (per partition), and no binary search style back and forth would be necessary. The output could e.g. be

  • lag < 10
  • lag < 100
  • lag < 1000
  • lag < 10000
  • ...
user_7095370
Related