@Override
        public Void get() throws ExecutionException {
            final AtomicReference<Throwable> firstError = new AtomicReference<>(null);
            try {
                removeTopologyFuture.get();
            } catch (final InterruptedException | ExecutionException e) {
                final Throwable error = e.getCause() != null ? e.getCause() : e;
                log.error("Removing named topology failed. Offset reset will still be attempted.", error);
                firstError.compareAndSet(e, null);
            }
            try {
                resetOffsets.run();
            } catch (final Throwable e) {
                log.error("Failed to reset offsets, you should do so manually if you want to add new topologies"
                              + "in the future that consume from the same input topics");
                firstError.compareAndSet(e, null);
            }

            if (firstError.get() != null) {
                throw new ExecutionException(firstError.get());
            } else {
                return null;
            }
        }
@return a {@link KafkaFuture} that completes successfully when all threads on this client have removed the
corresponding {@link NamedTopology} and all source topic offsets have been deleted (if applicable). At this
point no more of its tasks will be processed by the current client, but there may still be other clients which
do. It is only guaranteed that this {@link NamedTopology} has fully stopped processing when all clients have
successfully completed their corresponding {@link KafkaFuture}.
streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/RemoveNamedTopologyResult.java