Page History

public class CloseOptions {
    /**
     * Enum to specify the group membership operation upon leaving group.
     *
     * <ul>
     *   <li><b>{@code LEAVE_GROUP}</b>: means the consumer leave the group.</li>
     *   <li><b>{@code REMAIN_IN_GROUP}</b>: means the consumer will remain in the group.</li>
     * </ul>
     */
    public enum GroupMembershipOperation {
        LEAVE_GROUP,
        REMAIN_IN_GROUP
    }

    /**
     * Specifies the group membership operation upon shutdown.
     * By default, {@code GroupMembershipOperation.REMAIN_IN_GROUP} will be applied, which follows the KafkaStreams default behavior.
     */
    protected GroupMembershipOperation operation = GroupMembershipOperation.REMAIN_IN_GROUP;

    /**
     * Specifies the maximum amount of time to wait for the close process to complete.
     * This allows users to define a custom timeout for gracefully stopping the KafkaStreams.
     */
    protected Optional<Duration> timeout = Optional.of(Duration.ofMillis(Long.MAX_VALUE));

    private CloseOptions() {
    }

    protected CloseOptions(final CloseOptions CloseOptions) {
        this.operation = CloseOptions.operation;
        this.timeout = CloseOptions.timeout;
    }

    /**
     * Static method to create a {@code CloseOptions} with a custom timeout.
     *
     * @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;
    }
}