KStream-KStream Join not trigger left-join

I utilize kafka streams together with spring-cloud-stream-binder-kafka-streams to build a stream that receives from a topic information about pods running in a kubernetes cluster and build out of it a mapping of container image with associated pods.

public class ContainerPodMap {
private String image;
private List<String> pods;
}

Based on this I try to join a kstream that contains scan actions. This join gets triggered once properly, but as soon as I add new entries via producing into the running pods topic the leftjoin action gets never triggered again even though I would expect that a new pod event with a new containerpod-map should create a leftjoin where the scanAction is empty.

I implemented some peeks to see what is happening and it looks like the new entries simply "dies" infront of the leftjoin. So the peek container-map: <key> - <value> works as expected.

@Bean
    public BiFunction<KStream<String, PodEventDto>, KStream<String, ScanAction>, KStream<String, ScanAction>> events() {
        return (podEventStream, scanActionKStream) -> {
            return podEventStream
                    .process(PodEventStreamProcessor::new, "container-pod-map")
                    .peek((key, value) -> {
                        System.out.println("container-map: " + key + " - " + value.getImage());
                    })
                    .map((key, value) -> KeyValue.pair(key, value))
                    .leftJoin(scanActionKStream,
                             (key, containerPodMap, scanAction) -> {
                                /// never arrive here after the initial startup
                             },
                              JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(SCAN_INTERVAL_IN_MINUTES)),
                             StreamJoined.with(Serdes.String(), new JsonSerde<>(ContainerPodMap.class), new JsonSerde<>(ScanAction.class)))
    ///...
    }

public static class PodEventStreamProcessor extends ContextualProcessor<String, PodEventDto, String, ContainerPodMap> {

        private KeyValueStore<String, ContainerPodMap> stateStore;

        @Override
        public void init(ProcessorContext<String, ContainerPodMap> context) {
            super.init(context);
            stateStore = context.getStateStore("container-pod-map");
        }
    
        @Override
        public void process(Record<String, PodEventDto> podEventDtoRecord) {
            String podId = podEventDtoRecord.value().getId();
            for (PodContainerEventDto podContainer : podEventDtoRecord.value().getContainer()) {
                // simplyfied to reduce complexity in example
                context().forward(stringContainerPodMapRecord);
            }

                     

spring.cloud.stream.bindings.events-in-0.destination=pod-event
spring.cloud.stream.bindings.events-in-1.destination=scan-tasks
spring.cloud.stream.bindings.events-out-0.destination=scan-tasks
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.springframework.kafka.support.serializer.JsonSerde
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde

Left join results are only emitted after the join window was closed. As long as the join window is open, it's not clear if the left input record will produce an inner join result or not. Only if the left input record does not produce an inner join result, it would be emitted as left join result.

If you only send a single record, stream time (which is only advanced based on record timestamps) does not advance, and thus the join window is not closed. Only if another record with a large enough timestamp is processed, such that the join window is close, a left-join result would be emitted.

Cf https://www.confluent.io/events/kafka-summit-europe-2021/temporal-joins-in-kafka-streams-and-ksqldb/