Ahh... That makes a lot of sense now. Thanks guys. :)
So to fix this, I have to either adjust all my own time units (except source partition timeout) to values of unit seconds, or use milliseconds and adjust the measurement timestamps accordingly? That explains now, why I didn't find many information about which time unit to use for the input parameters, if it depends on the input data timestamp unit.
It is possible that bigger lags are occurring within the measurement stream, but that is not a big deal. They are supposed to be discarded as the events are too late to be processed anyway.
traverseIterable(map.entrySet()).map(e -> Util.entry(e.getKey(), e.getValue());
return traverseIterable(correlationIds .entrySet().stream() .map(e -> entry(e.getKey(), e.getValue())) .collect(Collectors.toList()));