He escrito este código en una aplicación de transmisión de 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 debería (si lo entendí correctamente) emitir registros por Clave después de cerrar la ventana. De alguna manera, el comportamiento es el siguiente:

La transmisión no emite el primer registro y solo lo reenvía después del segundo registro, incluso con una clave diferente, y luego el segundo registro se emite solo después del tercero y así sucesivamente ..

He intentado múltiples StreamConfigs con "exactamente_una vez" y con o sin almacenamiento en caché también, este comportamiento persiste.

Gracias de antemano por tu ayuda