private void resetOffsets(final Set<TopicPartition> partitionsToReset) throws StreamsException {
        // The number of times to retry upon failure
        int retries = 100;
        while (true) {
            try {
                final DeleteConsumerGroupOffsetsResult deleteOffsetsResult = adminClient.deleteConsumerGroupOffsets(
                    applicationConfigs.getString(StreamsConfig.APPLICATION_ID_CONFIG),
                    partitionsToReset);
                deleteOffsetsResult.all().get();
                log.info("Successfully completed resetting offsets.");
                break;
            } catch (final InterruptedException ex) {
                ex.printStackTrace();
                log.error("Offset reset failed.", ex);
                throw new StreamsException(ex);
            } catch (final ExecutionException ex) {
                final Throwable error = ex.getCause() != null ? ex.getCause() : ex;

                if (error instanceof GroupSubscribedToTopicException &&
                    error.getMessage()
                        .equals("Deleting offsets of a topic is forbidden while the consumer group is actively subscribed to it.")) {
                    log.debug("Offset reset failed, there may be other nodes which have not yet finished removing this topology", error);
                } else if (error instanceof GroupIdNotFoundException) {
                    log.info("The offsets have been reset by another client or the group has been deleted, no need to retry further.");
                    break;
                } else {
                    if (--retries > 0) {
                        log.error("Offset reset failed, retries remaining: " + retries, error);
                    } else {
                        log.error("Offset reset failed, no retries remaining.", error);
                        throw new StreamsException(error);
                    }
                }
            }
            try {
                Thread.sleep(100);
            } catch (final InterruptedException ex) {
                ex.printStackTrace();
            }
        }
    }
@return  true iff the application is still in CREATED and the future was completed
streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java