@Override
        public synchronized void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null) {
                if (!completed) {
                    this.exception = exception;
                    callback.onCompletion(exception, null);
                    completed = true;
                    this.notify();
                }
                return;
            }

            numLeft -= 1;
            if (numLeft == 0) {
                callback.onCompletion(null, null);
                completed = true;
                this.notify();
            }
        }
Stop reading from and writing to the offsets topic, and relinquish resources allocated for interacting
with it, including Kafka clients.
<p>
The admin client derived from the given {@link Supplier} will not be closed and it is the
caller's responsibility to manage its lifecycle accordingly.
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java