Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java (Java)

Code before:
    }

    public static final AutoOffsetResetStrategy EARLIEST = new AutoOffsetResetStrategy(StrategyType.EARLIEST);
    public static final AutoOffsetResetStrategy LATEST = new AutoOffsetResetStrategy(StrategyType.LATEST);
    public static final AutoOffsetResetStrategy NONE = new AutoOffsetResetStrategy(StrategyType.NONE);

    private final StrategyType type;
    private final Optional<Duration> duration;

    private AutoOffsetResetStrategy(StrategyType type) {
        this(type, Optional.empty());
    }

    private AutoOffsetResetStrategy(StrategyType type, Optional<Duration> duration) {
        this.type = type;
        this.duration = duration;
    }

    /**
     * Returns true if the given string is a valid auto offset reset strategy.
     */
    public static boolean isValid(String offsetStrategy) {
        if (offsetStrategy == null) {
            return false;
        } else if (StrategyType.BY_DURATION.toString().equals(offsetStrategy)) {
            return false;
        } else if (Arrays.asList(Utils.enumOptions(StrategyType.class)).contains(offsetStrategy)) {
            return true;
        } else if (offsetStrategy.startsWith(StrategyType.BY_DURATION + ":")) {
            String isoDuration = offsetStrategy.substring(StrategyType.BY_DURATION.toString().length() + 1);
            try {
                Duration duration = Duration.parse(isoDuration);
                return !duration.isNegative();
            } catch (Exception e) {
                return false;
            }
        } else {
            return false;
        }
    }

    /**
     *  Returns the AutoOffsetResetStrategy from the given string.
     */
    public static AutoOffsetResetStrategy fromString(String offsetStrategy) {
        if (!isValid(offsetStrategy)) {
            throw new IllegalArgumentException("Unknown auto offset reset strategy: " + offsetStrategy);
        }

        if (offsetStrategy.startsWith(StrategyType.BY_DURATION + ":")) {
            String isoDuration = offsetStrategy.substring(StrategyType.BY_DURATION.toString().length() + 1);

Review comment (bug):
should we add a log.warn for this missing duration? (or propagate the error to give better visibility on why the duration config is not ok? I expect this might be a very common mistake initially.

With the current shape seems that any wrong value in this config will end up just showing a generic msg IllegalArgumentException("Unknown auto offset reset strategy: "..) right? 

Code after:
    }

    public static final AutoOffsetResetStrategy EARLIEST = new AutoOffsetResetStrategy(StrategyType.EARLIEST);
    public static final AutoOffsetResetStrategy LATEST = new AutoOffsetResetStrategy(StrategyType.LATEST);
    public static final AutoOffsetResetStrategy NONE = new AutoOffsetResetStrategy(StrategyType.NONE);

    private final StrategyType type;
    private final Optional<Duration> duration;

    private AutoOffsetResetStrategy(StrategyType type) {
        this.type = type;
        this.duration = Optional.empty();
    }

    private AutoOffsetResetStrategy(Duration duration) {
        this.type = StrategyType.BY_DURATION;
        this.duration = Optional.of(duration);
    }

    /**
     *  Returns the AutoOffsetResetStrategy from the given string.
     */
    public static AutoOffsetResetStrategy fromString(String offsetStrategy) {
        if (offsetStrategy == null) {
            throw new IllegalArgumentException("Auto offset reset strategy is null");
        }

        if (StrategyType.BY_DURATION.toString().equals(offsetStrategy)) {
            throw new IllegalArgumentException("<:duration> part is missing in by_duration auto offset reset strategy.");
        }

        if (Arrays.asList(Utils.enumOptions(StrategyType.class)).contains(offsetStrategy)) {
            StrategyType type = StrategyType.valueOf(offsetStrategy.toUpperCase(Locale.ROOT));
            switch (type) {
                case EARLIEST:
                    return EARLIEST;
                case LATEST:
                    return LATEST;
                case NONE:
                    return NONE;
                default:
                    throw new IllegalArgumentException("Unknown auto offset reset strategy: " + offsetStrategy);
            }
        }

        if (offsetStrategy.startsWith(StrategyType.BY_DURATION + ":")) {
            String isoDuration = offsetStrategy.substring(StrategyType.BY_DURATION.toString().length() + 1);
            try {
                Duration duration = Duration.parse(isoDuration);
                if (duration.isNegative()) {
                    throw new IllegalArgumentException("Negative duration is not supported in by_duration offset reset strategy.");