Kafka Streams topology with windowing doesn't trigger state changes

I am building the following Kafka Streams topology (pseudo code):

gK = builder.stream().gropuByKey();
g1 = gK.windowedBy(TimeWindows.of("PT1H")).reduce().mapValues().toStream().mapValues().selectKey();
g2 = gK.reduce().mapValues();
g1.leftJoin(g2).to();

If you notice, this is a rhomb-like topology that starts at single input topic and ends in the single output topic with messages flowing through two parallel flows that eventually get joined together at the end. One flow applies (tumbling?) windowing, the other does not. Both parts of the flow work on the same key (apart from the WindowedKey intermediately introduced by the windowing).

The timestamp for my messages is event-time. That is, they get picked from the message body by my custom configured TimestampExtractor implementation. The actual timestamps in my messages are several years to the past.

That all works well at first sight in my unit tests with a couple of input/output messages and in the runtime environment (with real Kafka).

The problem seems to come when the number of messages starts being significant (e.g. 40K).

My failing scenario is following:

1. ~40K records with the same key get uploaded into the input topic first

2. ~40K updates are coming out of the output topic, as expected

3. another ~40K records with the same but different to step 1) key get uploaded into the input topic

4. only ~100 updates are coming out of the output topic, instead of expected new ~40K updates. There is nothing special to see on those ~100 updates, their contents seems to be right, but only for certain time windows. For other time windows there are no updates even though the flow logic and input data should definetly generate 40K records. In fact, when I exchange dataset in step 1) and 3) I have exactly same situation with ~40K updates coming from the second dataset and same number ~100 from the first.

I can easily reproduce this issue in the unit tests using TopologyTestDriver locally (but only on bigger numbers of input records).

In my tests, I've tried disabling caching with StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG. Unfortunately, that didn't make any difference.

UPDATE

I tried both, reduce() calls and aggregate() calls instead. The issue persists in both cases.

What I'm noticing else is that with StreamsConfig.TOPOLOGY_OPTIMIZATION set to StreamsConfig.OPTIMIZE and without it, the mapValues() handler gets called in debugger before the preceding reduce() (or aggregate()) handlers at least for the first time. I didn't expect that.

Tried both join() and leftJoin() unfortunately same result. In debugger the second portion of the data doesn't trigger reduce() handler in the "left" flow at all, but does trigger reduce() handler in the "right" flow.

With my configuration, if the number or records in both datasets is 100 in each, the problem doesn't manifests itself, I'm getting 200 output messages as I expect. When I raise the number to 200 in each data set, I'm getting less than 400 expected messages out. So, it seems at the moment that something like "old" windows get dropped and the new records for those old windows get ignored by the stream. There is window retention setting that can be set, but with its default value that I use I was expecting for windows to retain their state and stay active for at least 12 hours (what exceeds the time of my unit test run significantly).

Tried to amend the left reducer with the following Window storage config:

Materialized.as(
    Stores.inMemoryWindowStore(
        "rollup-left-reduce",
        Duration.ofDays(5 * 365),
        Duration.ofHours(1), false)
)

still no difference in results.

Same issue persists even with only single "left" flow without the "right" flow and without join(). It seems that the problem is in the window retention settings of my set up. Timestamps (event-time) of my input records span 2 years. The second dataset starts from the beginning of 2 years again. this place in Kafka Streams makes sure that the second data set records get ignored:

https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java#L125

Kafka Streams Version is 2.4.0. Also using Confluent dependencies version 5.4.0.

My questions are

• What could be the reason for such behaviour?
• Did I miss anything in my stream topology?
• Is such topology expected to work at all?