code: @Override
  	public void close() {
  		synchronized (lock) {
  			next = null;
  			wakeupProducer = false;
  
  			if (error == null) {
  				error = new ClosedException();
  			}
  			lock.notifyAll();
  		}
  	}
docs: Closes the handover. Both the {@link #produce(ConsumerRecords)} method and the
  {@link #pollNext()} will throw a {@link ClosedException} on any currently blocking and
  future invocations.
  
  <p>If an exception was previously reported via the {@link #reportError(Throwable)} method,
  that exception will not be overridden. The consumer thread will throw that exception upon
  calling {@link #pollNext()}, rather than the {@code ClosedException}.