@Override
	public void close() throws IOException {
		if (offsets != null) {
			shutdownLatch = new CountDownLatch(1);
			offsets.onCompleted();
			try {
				shutdownLatch.await(shutdownTimeout, TimeUnit.MILLISECONDS);
			}
			catch (InterruptedException e) {
				// ignore
			}
			subscription.unsubscribe();
		}
		delegate.close();
	}
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