private void setupStartupMode() {
    String startupMode =
        CompatiblePropertyUtil.propertyAsString(
                tableProperties, SCAN_STARTUP_MODE.key(), SCAN_STARTUP_MODE.defaultValue())
            .toLowerCase();

    switch (startupMode) {
      case SCAN_STARTUP_MODE_EARLIEST:
        setStartingOffsets(OffsetsInitializer.earliest());
        break;
      case SCAN_STARTUP_MODE_LATEST:
        setStartingOffsets(OffsetsInitializer.latest());
        break;
      case SCAN_STARTUP_MODE_TIMESTAMP:
        long startupTimestampMillis =
            Long.parseLong(
                Preconditions.checkNotNull(
                    tableProperties.get(SCAN_STARTUP_TIMESTAMP_MILLIS.key()),
                    String.format(
                        "'%s' should be set in '%s' mode",
                        SCAN_STARTUP_TIMESTAMP_MILLIS.key(), SCAN_STARTUP_MODE_TIMESTAMP)));
        setStartingOffsets(OffsetsInitializer.timestamp(startupTimestampMillis));
        break;
      case SCAN_STARTUP_MODE_GROUP_OFFSETS:
        setStartingOffsets(OffsetsInitializer.committedOffsets());
        break;
      case SCAN_STARTUP_MODE_SPECIFIC_OFFSETS:
        Map<TopicPartition, Long> specificOffsets = new HashMap<>();
        String specificOffsetsStrOpt =
            Preconditions.checkNotNull(
                tableProperties.get(SCAN_STARTUP_SPECIFIC_OFFSETS.key()),
                String.format(
                    "'%s' should be set in '%s' mode",
                    SCAN_STARTUP_SPECIFIC_OFFSETS.key(), SCAN_STARTUP_MODE_SPECIFIC_OFFSETS));
        final Map<Integer, Long> offsetMap =
            parseSpecificOffsets(specificOffsetsStrOpt, SCAN_STARTUP_SPECIFIC_OFFSETS.key());
        offsetMap.forEach(
            (partition, offset) -> {
              final TopicPartition topicPartition =
                  new TopicPartition(getLogTopic(tableProperties).get(0), partition);
              specificOffsets.put(topicPartition, offset);
            });
        setStartingOffsets(OffsetsInitializer.offsets(specificOffsets));
        break;
      default:
        throw new ValidationException(
            String.format(
                "%s only support '%s', '%s', '%s', '%s', '%s'. But input is '%s'",
                MixedFormatValidator.SCAN_STARTUP_MODE,
                SCAN_STARTUP_MODE_LATEST,
                SCAN_STARTUP_MODE_EARLIEST,
                SCAN_STARTUP_MODE_TIMESTAMP,
                SCAN_STARTUP_MODE_GROUP_OFFSETS,
                SCAN_STARTUP_MODE_SPECIFIC_OFFSETS,
                startupMode));
    }
  }
Build the {@link LogKafkaSource}.

@return a LogKafkaSource with the settings made for this builder.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/kafka/LogKafkaSourceBuilder.java