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

Code before:
     * @param timeout the maximum time to wait for the KafkaStreams to close.
     * @return a new {@code CloseOptions} instance with the specified timeout.
     */
    public static CloseOptions timeout(final Duration timeout) {
        return new CloseOptions().withTimeout(timeout);
    }

    /**
     * Static method to create a {@code CloseOptions} with a specified group membership operation.
     *
     * @param operation the group membership operation to apply. Must be one of {@code LEAVE_GROUP}, {@code REMAIN_IN_GROUP}.
     * @return a new {@code CloseOptions} instance with the specified group membership operation.
     */
    public static CloseOptions groupMembershipOperation(final GroupMembershipOperation operation) {
        return new CloseOptions().withGroupMembershipOperation(operation);
    }

    /**
     * Fluent method to set the timeout for the close process.
     *
     * @param timeout the maximum time to wait for the KafkaStreams to close. If {@code null}, the default timeout will be used.
     * @return this {@code CloseOptions} instance.
     */
    public CloseOptions withTimeout(final Duration timeout) {
        if (timeout == null) {
            this.timeout = Optional.of(Duration.ofMillis(Long.MAX_VALUE));
        } else {
            this.timeout = Optional.of(timeout);
        }
        return this;
    }

    /**
     * Fluent method to set the group membership operation upon shutdown.
     *
     * @param operation the group membership operation to apply. Must be one of {@code LEAVE_GROUP}, {@code REMAIN_IN_GROUP}.
     * @return this {@code CloseOptions} instance.
     */
    public CloseOptions withGroupMembershipOperation(final GroupMembershipOperation operation) {
        this.operation = Objects.requireNonNull(operation, "operation should not be null");
        return this;
    }
}


Review comment (bug):
It looks like `timeout` is never `empty`. However, I think keeping the `empty` state is useful because it shows the user did not explicitly set a `timeout`. Maybe we can use `orElse` to address the comments

```java
        var timeout = optionsInternal.timeout().orElse(Duration.ofMillis(Long.MAX_VALUE));
        final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout, "timeout");
        final long timeoutMs = validateMillisecondDuration(timeout, msgPrefix);
```

Code after:
     * @param timeout the maximum time to wait for the KafkaStreams to close.
     * @return a new {@code CloseOptions} instance with the specified timeout.
     */
    public static CloseOptions timeout(final Duration timeout) {
        return new CloseOptions().withTimeout(timeout);
    }

    /**
     * Static method to create a {@code CloseOptions} with a specified group membership operation.
     *
     * @param operation the group membership operation to apply. Must be one of {@code LEAVE_GROUP}, {@code REMAIN_IN_GROUP}.
     * @return a new {@code CloseOptions} instance with the specified group membership operation.
     */
    public static CloseOptions groupMembershipOperation(final GroupMembershipOperation operation) {
        return new CloseOptions().withGroupMembershipOperation(operation);
    }

    /**
     * Fluent method to set the timeout for the close process.
     *
     * @param timeout the maximum time to wait for the KafkaStreams to close. If {@code null}, the default timeout will be used.
     * @return this {@code CloseOptions} instance.
     */
    public CloseOptions withTimeout(final Duration timeout) {
        this.timeout = Optional.ofNullable(timeout);
        return this;
    }

    /**
     * Fluent method to set the group membership operation upon shutdown.
     *
     * @param operation the group membership operation to apply. Must be one of {@code LEAVE_GROUP}, {@code REMAIN_IN_GROUP}.
     * @return this {@code CloseOptions} instance.
     */
    public CloseOptions withGroupMembershipOperation(final GroupMembershipOperation operation) {
        this.operation = Objects.requireNonNull(operation, "operation should not be null");
        return this;
    }
}