InvalidTimestampException when run emit changes queries

[onepas]

If messages retain in ksql_fraud_whitelist topic > 5 days, I got InvalidTimestampException when run emit changes queries.

ksql> select * from ksql_fraud_whitelist;
+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+
|ID            |NAME          |ACCOUNT_TYPE  |CIF           |GENDER        |DATE_OF_BIRTH |NID           |DATE_OF_ISSUE |PLACE_OF_ISSUE|ADDRESS       |ACTIVE        |
+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+
|111222        |Minh001       |Personal      |null          |N/A           |null          |null          |null          |null          |null          |1             |
|222333        |Minh002       |Business      |null          |Male          |null          |null          |null          |null          |null          |1             |
|333444        |Minh003       |Business      |null          |Male          |null          |null          |null          |null          |null          |1             |
Query terminated

ksql> select * from ksql_fraud_whitelist emit changes;
+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+
|ID            |NAME          |ACCOUNT_TYPE  |CIF           |GENDER        |DATE_OF_BIRTH |NID           |DATE_OF_ISSUE |PLACE_OF_ISSUE|ADDRESS       |ACTIVE        |
+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+--------------+
|111222        |Minh001       |Personal      |null          |N/A           |null          |null          |null          |null          |null          |1             |
org.apache.kafka.streams.errors.StreamsException: Error encountered sending record to topic _confluent-ksql-default_transient_transient_KSQL_FRAUD_WHITELIST_7880574491299092471_1702953261790-KsqlTopic-Reduce-changelog for task 0_0 due to:
org.apache.kafka.common.errors.InvalidTimestampException: Timestamp 1702458858953 of message with offset 0 is out of range. The timestamp should be within [1702866870161, 1703039670161]
Exception handler choose to FAIL the processing, no more records would be sent.

[GMaystorski]

You should configure this property for the topic in question - message.timestamp.difference.max.ms . Or set broker defaults that are very large to not encounter this issue for autocreated topics

[mjsax]

What @GMaystorski says.

When you run the first query, no data is written into a topic, and thus the query works (the query is directly answered over data in RocksDB state stores). -- For the "emit changes" query, ksqlDB executes the query differently. But the error is not related to ksqlDB itself, but it's a Kafka configuration thing.