@Override
	public void afterPropertiesSet() throws Exception {
		Assert.isTrue(timespan > 0 ^ count > 0, "Only one of the timespan or count must be set");
		// create the stream if windowing is set, and count is higher than 1
		if (timespan > 0 || count > 1) {
			offsets = new SerializedSubject<>(PublishSubject.<PartitionAndOffset>create());
			// window by either count or time
			Observable<Observable<PartitionAndOffset>> window =
					timespan > 0 ? offsets.window(timespan, TimeUnit.MILLISECONDS) : offsets.window(count);
			Observable<PartitionAndOffset> maximumOffsetsByWindow = window
					.flatMap(findHighestOffsetsByPartition)
					.doOnCompleted(notifyObservableClosed);
			subscription = maximumOffsetsByWindow.subscribe(delegateUpdateOffsetAction);
		}
		else {
			offsets = null;
		}
	}
The timeout that {@link #close()} and {@link #destroy()} operations will wait for receving a confirmation that the
underlying writes have been processed.

@param shutdownTimeout duration in milliseconds
extensions/spring-xd-extension-kafka/src/main/java/org/springframework/integration/x/kafka/WindowingOffsetManager.java