Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java (Java)

Code before:
    /**
     * Shutdown this {@code KafkaStreams} by signaling all the threads to stop, and then wait up to the timeout for the
     * threads to join.
     * This method is deprecated and replaced by {@link #close(org.apache.kafka.streams.CloseOptions)}.
     * @param options  contains timeout to specify how long to wait for the threads to shut down, 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}
     */
    @Deprecated(since = "4.2")
    public synchronized boolean close(final CloseOptions options) throws IllegalArgumentException {
        final org.apache.kafka.streams.CloseOptions closeOptions = org.apache.kafka.streams.CloseOptions.timeout(options.timeout)
                .withGroupMembershipOperation(options.leaveGroup ?
                        org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP :
                        org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP);
        return close(closeOptions);
    }

    /**
     * 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 shut down, and a flag
     *                 {@link org.apache.kafka.streams.CloseOptions.GroupMembershipOperation#LEAVE_GROUP} 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}
     */
    public synchronized boolean close(final org.apache.kafka.streams.CloseOptions options) throws IllegalArgumentException {
        Objects.requireNonNull(options, "options cannot be null");
        final CloseOptionsInternal optionsInternal = new CloseOptionsInternal(options);
        final String msgPrefix = prepareMillisCheckFailMsgPrefix(optionsInternal.timeout(), "timeout");
        final long timeoutMs = validateMillisecondDuration(optionsInternal.timeout().get(), msgPrefix);
        if (timeoutMs < 0) {
            throw new IllegalArgumentException("Timeout can't be negative.");
        }

        return close(Optional.of(timeoutMs), optionsInternal.operation());
    }

    /**
     * Do a cleanup of the local {@link StateStore} directory ({@link StreamsConfig#STATE_DIR_CONFIG}) by deleting all
     * data with regard to the {@link StreamsConfig#APPLICATION_ID_CONFIG application ID}.
     * <p>
     * May only be called either before this {@code KafkaStreams} instance is {@link #start() started} or after the
     * instance is {@link #close() closed}.
     * <p>
     * Calling this method triggers a restore of local {@link StateStore}s on the next {@link #start() application start}.

Review comment (suggestion):
Shouldn't this just specify that users can provide either option for remaining in the group or leaving vs. explicity showing `LEAVE_GROUP`

Code after:
    /**
     * Shutdown this {@code KafkaStreams} by signaling all the threads to stop, and then wait up to the timeout for the
     * threads to join.
     * This method is deprecated and replaced by {@link #close(org.apache.kafka.streams.CloseOptions)}.
     * @param options  contains timeout to specify how long to wait for the threads to shut down, 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}
     */
    @Deprecated(since = "4.2")
    public synchronized boolean close(final CloseOptions options) throws IllegalArgumentException {
        final org.apache.kafka.streams.CloseOptions closeOptions = org.apache.kafka.streams.CloseOptions.timeout(options.timeout)
                .withGroupMembershipOperation(options.leaveGroup ?
                        org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP :
                        org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP);
        return close(closeOptions);
    }

    /**
     * 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 shut down,
     *                 and a {@link org.apache.kafka.streams.CloseOptions.GroupMembershipOperation}
     *                 to trigger consumer leave call or remain in the group
     * @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}
     */
    public synchronized boolean close(final org.apache.kafka.streams.CloseOptions options) throws IllegalArgumentException {
        Objects.requireNonNull(options, "options cannot be null");
        final CloseOptionsInternal optionsInternal = new CloseOptionsInternal(options);
        final String msgPrefix = prepareMillisCheckFailMsgPrefix(optionsInternal.timeout(), "timeout");
        final long timeoutMs = validateMillisecondDuration(optionsInternal.timeout().get(), msgPrefix);
        if (timeoutMs < 0) {
            throw new IllegalArgumentException("Timeout can't be negative.");
        }

        return close(Optional.of(timeoutMs), optionsInternal.operation());
    }

    /**
     * Do a cleanup of the local {@link StateStore} directory ({@link StreamsConfig#STATE_DIR_CONFIG}) by deleting all
     * data with regard to the {@link StreamsConfig#APPLICATION_ID_CONFIG application ID}.
     * <p>
     * May only be called either before this {@code KafkaStreams} instance is {@link #start() started} or after the
     * instance is {@link #close() closed}.
     * <p>
     * Calling this method triggers a restore of local {@link StateStore}s on the next {@link #start() application start}.