How to specify retention period of join window?

user_1992501 0
Viewed 1397

I want to join two streams and I have set the join window to 25 hours as the records to be joined can be a maximum of 24 hours apart. 

final Long JOIN_WINDOW = TimeUnit.HOURS.toMillis(25);

kstream.join(
  runsheetIdStream,
  (jt,r) -> { jt.setDate(r.getStart_date()); return jt; },
  JoinWindows.of(JOIN_WINDOW),
  Joined.with(Serdes.Long(),jobTransactionSerde,runsheetSerde))

This is throwing the following exception:

org.apache.kafka.streams.errors.TopologyException: Invalid topology: The retention period of the join window KSTREAM-JOINTHIS-0000000016-store must be no smaller than its window size.

How do I increase the retention period?

apache-kafka-streams apache-kafka
1 Answers

When you join and used JoinWindows.of(JOIN_WINDOW) you implicitly defined the metadata of the underlying state store.

From the javadoc of JoinWindows.of:

Specifies that records of the same key are joinable if their timestamps are within timeDifference, i.e., the timestamp of a record from the secondary stream is max timeDifference earlier or later than the timestamp of the record from the primary stream.

The so-called retention period (aka window maintain duration) was earlier (before Kafka Streams 2.1.0) specified using until:

Set the window maintain duration (retention time) in milliseconds. This retention time is a guaranteed lower bound for how long a window will be maintained.

Since by default the retention is 1 day (can't find the reference at the moment) that's the reason for the exception.

As of Kafka Streams 2.1.0 you should be using Materialized API:

Used to describe how a StateStore should be materialized. You can either provide a custom StateStore backend through one of the provided methods accepting a supplier or use the default RocksDB backends by providing just a store name.

Materialized gives you a full control over the underlying state store for join and gives withRetention(java.time.Duration retention):

Configure retention period for window and session stores.

Note that the retention period must be at least long enough to contain the windowed data's entire life cycle, from window-start through window-end, and for the entire grace period.

user_13053440
Related