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

Code before:
     * @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 Duration timeout = optionsInternal.timeout().orElse(Duration.ofMillis(Long.MAX_VALUE));
        final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout, "timeout");
        final long timeoutMs = validateMillisecondDuration(timeout, 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}.
     *
     * @throws IllegalStateException if this {@code KafkaStreams} instance has been started and hasn't fully shut down
     * @throws StreamsException if cleanup failed
     */
    public void cleanUp() {
        if (!(state.hasNotStarted() || state.hasCompletedShutdown())) {
            throw new IllegalStateException("Cannot clean up while running.");
        }

Review comment (suggestion):
Perhaps using `orElseGet` is more better, which can avoid creating `Duration`

Code after:
     * 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 Duration timeout = optionsInternal.timeout().orElseGet(() -> Duration.ofMillis(Long.MAX_VALUE));
        final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout, "timeout");
        final long timeoutMs = validateMillisecondDuration(timeout, 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}.
     *
     * @throws IllegalStateException if this {@code KafkaStreams} instance has been started and hasn't fully shut down
     * @throws StreamsException if cleanup failed
     */
    public void cleanUp() {
        if (!(state.hasNotStarted() || state.hasCompletedShutdown())) {
            throw new IllegalStateException("Cannot clean up while running.");