private OffsetResetStrategy getResetStrategy(String offsetResetConfig) {
        return Arrays.stream(OffsetResetStrategy.values())
                .filter(ors -> ors.name().equals(offsetResetConfig.toUpperCase(Locale.ROOT)))
                .findAny()
                .orElseThrow(
                        () -> new IllegalArgumentException(
                                String.format(
                                        "%s can not be set to %s. Valid values: [%s]",
                                        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
                                        offsetResetConfig,
                                        Arrays.stream(OffsetResetStrategy.values())
                                                .map(Enum::name)
                                                .map(String::toLowerCase)
                                                .collect(Collectors.joining(",")))));
    }
Flag to determine source mode. In upsert mode, it will keep the tombstone message. *
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/table/KafkaDynamicSource.java