How to send final kafka-streams aggregation result of a time windowed KTable?

by

In Kafka Streams there is no such thing as a “final aggregation”. Windows are kept open all the time to handle out-of-order records that arrive after the window end-time passed. However, windows are not kept forever. They get discarded once their retention time expires. There is no special action as to when a window gets discarded.

See Confluent documentation for more details: http://docs.confluent.io/current/streams/

Thus, for each update to an aggregation, a result record is produced (because Kafka Streams also update the aggregation result on out-of-order records). Your “final result” would be the latest result record (before a window gets discarded). Depending on your use case, manual de-duplication would be a way to resolve the issue (using lower lever API, transform() or process())

This blog post might help, too: https://timothyrenner.github.io/engineering/2016/08/11/kafka-streams-not-looking-at-facebook.html

Another blog post addressing this issue without using punctuations: http://blog.inovatrend.com/2018/03/making-of-message-gateway-with-kafka.html

Update

With KIP-328, a KTable#suppress() operator is added, that will allow to suppress consecutive updates in a strict manner and to emit a single result record per window; the tradeoff is an increase latency.

More Related Contents:

  1. How to commit manually with Kafka Stream?
  2. Kafka Stream Suppress session-windowed-aggregation
  3. Streamparse wordcount example
  4. Difference between session.timeout.ms and max.poll.interval.ms for Kafka >= 0.10.1
  5. Is there a way to purge the topic in Kafka?
  6. How does an offset expire for an Apache Kafka consumer group?
  7. Increase the number of messages read by a Kafka consumer in a single poll
  8. Kafka how to read from __consumer_offsets topic
  9. How to change the number of replicas of a Kafka topic?
  10. Delete message after consuming it in KAFKA
  11. Understanding Kafka Topics and Partitions
  12. When to use RabbitMQ over Kafka? [closed]
  13. Delete topic in Kafka 0.8.1.1
  14. Handling exceptions in Kafka streams
  15. Is it possible to access message headers with Kafka Streams?
  16. Dynamically connecting a Kafka input stream to multiple output streams
  17. Use Confluent Hub without Confluent Platform installation
  18. How to change start offset for topic?
  19. Kafka 0.11 how to reset offsets
  20. How can I send large messages with Kafka (over 15MB)?
  21. Using Kafka as a (CQRS) Eventstore. Good idea?
  22. How Can we create a topic in Kafka from the IDE using API
  23. What determines Kafka consumer offset?
  24. How to write spark streaming DF to Kafka topic
  25. Writing Custom Kafka Serializer
  26. Kafka server configuration – listeners vs. advertised.listeners
  27. When does the Apache Kafka client throw a “Batch Expired” exception?
  28. What is a simple, effective way to debug custom Kafka connectors?
  29. How to integrate Oracle and Kafka
  30. How can I delete a topic in Apache Kafka? [duplicate]

Leave a Comment