kafka stream heap usage

I am using kafka stream 2.1

I am trying to aggregate a stream of messages by their id. We have approximatively 20 messages with same id produced almost at the same time (a couple hundred ms maximum between two messages). So I am using a session window with a inactivity gap of 500ms and grace time of 5 seconds.

The incoming records has the ID as a key, and the value is made of few string fields + a map that can contains from 0 to few hundreds of entry (key is a string, value is an object with one string field).

Here is the code :

private final Duration INACTIVITY_GAP = Duration.ofMillis(500);
private final Duration GRACE_TIME = Duration.ofMillis(5000);

KStream<String, MyCustomMessage> source = streamsBuilder.stream("inputTopic", Consumed.with(Serdes.String(), myCustomSerde));

source
  .groupByKey(Grouped.with(Serdes.String(), myCustomSerde))                
  .windowedBy(SessionWindows.with(INACTIVITY_GAP).grace(GRACE_TIME))
  .aggregate(
     // initializer
     () -> {
         return new CustomAggMessage();
     },
     // aggregates records in same session
     (s, message, aggMessage) -> {
        // ...
         return aggMessage;
     },
     // merge sessions
     (s, aggMessage1, aggMessage2) -> {
         // ... merge
         return aggMessage2;
     },
     Materialized.with(Serdes.String(), myCustomAggSerde)
   )
   .suppress(Suppressed.untilWindowCloses(unbounded()))
   .selectKey((windowed, o) -> windowed.key());
   .toStream().to("outputTopic")

I also tried another Suppressed : .suppress(Suppressed.untilTimeLimit(Duration.ofSeconds(30), maxBytes(1_000_000_000L).emitEarlyWhenFull())). It did not help.

I have a custom rocksdb config :

public class CustomRocksDbConfig implements RocksDBConfigSetter {

    @Override
    public void setConfig(final String storeName, final Options options, final Map<String, Object> configs) {
        BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig();
        tableConfig.setCacheIndexAndFilterBlocks(true);
        options.setTableFormatConfig(tableConfig);
    }
}

The input topic has 32 partitions (about 10k msg/second) and we run 8 instances with 4 stream threads each.

When running this. The heap usage is very high (max heap is set to 4G, machine has 8G) and makes the app crash and restart, so the lag increases.

Does someone know why ? What could I change to make this working ? Is the session window and its parameters the right way to achieve this ?