private Consumer<StreamThread> streamThreadLeaveConsumerGroup(final long remainingTimeMs) {
        return thread -> {
            final Optional<String> groupInstanceId = thread.getGroupInstanceID();
            if (groupInstanceId.isPresent()) {
                log.debug("Sending leave group trigger to removing instance from consumer group: {}.",
                    groupInstanceId.get());
                final MemberToRemove memberToRemove = new MemberToRemove(groupInstanceId.get());
                final Collection<MemberToRemove> membersToRemove = Collections.singletonList(memberToRemove);

                final RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroupResult = adminClient
                    .removeMembersFromConsumerGroup(
                        applicationConfigs.getString(StreamsConfig.APPLICATION_ID_CONFIG),
                        new RemoveMembersFromConsumerGroupOptions(membersToRemove)
                    );

                try {
                    removeMembersFromConsumerGroupResult.memberResult(memberToRemove)
                        .get(remainingTimeMs, TimeUnit.MILLISECONDS);
                } catch (final Exception e) {
                    final String msg = String.format("Could not remove static member %s from consumer group %s.",
                                                     groupInstanceId.get(), applicationConfigs.getString(StreamsConfig.APPLICATION_ID_CONFIG));
                    log.error(msg, e);
                }
            }
        };
    }
Shutdown this {@code KafkaStreams} by signaling all the threads to stop, and then wait up to the timeout for the
threads to join.
@param options  contains timeout to specify how long to wait for the threads to shutdown, and a flag leaveGroup to
                trigger consumer leave call
@return {@code true} if all threads were successfully stopped&mdash;{@code false} if the timeout was reached
before all threads stopped
Note that this method must not be called in the {@link StateListener#onChange(KafkaStreams.State, KafkaStreams.State)} callback of {@link StateListener}.
@throws IllegalArgumentException if {@code timeout} can't be represented as {@code long milliseconds}
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java