# Kafka Session Windowing Suppress

**URL:** <https://forum.confluent.io/t/kafka-session-windowing-suppress/10657>\
**Category:** Kafka Streams\
**Created:** [9 May 2024 06:34 UTC](https://forum.confluent.io/t/kafka-session-windowing-suppress/10657 "2024-05-09T06:34:58Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![sangames\_k](https://avatars.discourse-cdn.com/v4/letter/s/b5ac83/32.png) [@sangames\_k](https://forum.confluent.io/u/sangames_k)\
**Post date:** [9 May 2024 06:34 UTC](https://forum.confluent.io/t/kafka-session-windowing-suppress/10657/1 "2024-05-09T06:34:58Z")

</div>

```auto
KTable<Windowed<String>, MyClass> windowedStream = sessionKStream
                .selectKey((k, v) -> v.getId())
                .groupByKey()
                .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(2)))
                .reduce((v1, v2) -> v2, Materialized.with(Serdes.String(), new MyClass()));
        KTable<Windowed<String>, MyClass> suppressedTable = windowedStream.suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()));
        KStream<String, MyClass> processedStream = suppressedTable.toStream().map((k, v) -> KeyValue.pair(k.key(), v));
                processedStream.transformValues(() -> new ValueTransformer<>() {
                    @Override
                    public void init(ProcessorContext context) {
                    }

                    @Override
                    public MyClass transform(MyClass value) {
                        if (value.getStatus() != null && !value.getStatus().equals("FINAL")) {
                            value.setStatus("FINAL");
                        }
                        return value;
                    }

                    @Override
                    public void close() {
                    }
                });
        processedStream.to("mytopic", Produced.with(Serdes.String(), new MyClass()));

```

At the end of window time of 2 minutes, the messages are not immediately produced to “mytopic”. Only if 4-5 messages ate produced in the input topic, only then a batch of messages are produced to the “mytopic”. The linger ms and batch size is also set to zero. But, still the messages are batched for some reason.

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [9 May 2024 17:29 UTC](https://forum.confluent.io/t/kafka-session-windowing-suppress/10657/2 "2024-05-09T17:29:27Z")

</div>

`suppress()` works on “stream-time” not wall-clock time. “stream-time” is base on the input record timestamps, so only if data is flowing, and input records with larger timestamps are processed, “stream-time” advanced allowing `suppress()` to emit results.

---

<div class="post-metadata">

**Author:** ![sangames\_k](https://avatars.discourse-cdn.com/v4/letter/s/b5ac83/32.png) [@sangames\_k](https://forum.confluent.io/u/sangames_k)\
**Post date:** [10 May 2024 05:35 UTC](https://forum.confluent.io/t/kafka-session-windowing-suppress/10657/3 "2024-05-10T05:35:22Z")

</div>

Is there some way to override this behaviour and emit the results as soon as the window ends?

---

<div class="post-metadata">

**Author:** ![mjsax](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/mjsax/32/3113_2.png) [@mjsax](https://forum.confluent.io/u/mjsax)\
**Post date:** [10 May 2024 16:02 UTC](https://forum.confluent.io/t/kafka-session-windowing-suppress/10657/4 "2024-05-10T16:02:12Z")

</div>

> emit the results as soon as the window ends

Sessions are _event_ driven and thus that’s not possible. If you have window gap of 5 second, and you get input events with ts, 1,2,3 but not more events, you have a session [1,3] and stream-time is 3. Even if the next event comes in at ts=9 and thus [1,3] is the session that will be closed, at stream-time 3 we don’t know what the next event will be. If the next event would have a ts=7 instead of ts=9, the session would be extended to [1,7] and would not close. – Thus, we can only close the window when we see ts=9. Does this make sense? Otherwise we would require some magic power to look into the future.

Note though: advance of time is not based on keys… So if you have a session for key=A of [1,3], and a record with key=B and ts=9 comes it, it would also close the session of key A (assuming that A and B are in the same partition).