Escrevi esse código em um aplicativo de fluxo Kafka:

KGroupedStream<String, foo> groupedStream = stream.groupByKey();
groupedStream.windowedBy(
SessionWindows.with(Duration.ofSeconds(3)).grace(Duration.ofSeconds(3)))
    .aggregate(() -> {...})
    .suppress(Suppressed.untilWindowCloses(unbounded()))
    .toStream()...

que deve (se bem entendi) emitir registros por chave depois que a janela for fechada. De alguma forma, o comportamento é o seguinte:

O fluxo não emite o primeiro registro e o encaminha somente após o segundo registro, mesmo com uma chave diferente, e o segundo registro é emitido somente após o terceiro e assim por diante.

Tentei vários StreamConfigs com "exact_once" e com ou sem o cache também, esse comportamento persist

Agradeço antecipadamente por sua ajuda