private void parseAndSetRequiredProperties() {
    maybeOverride(
        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName(), true);
    maybeOverride(
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        ByteArrayDeserializer.class.getName(),
        true);
    maybeOverride(ConsumerConfig.GROUP_ID_CONFIG, "KafkaSource-" + new Random().nextLong(), false);
    maybeOverride(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false", false);
    maybeOverride(
        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
        startingOffsetsInitializer.getAutoOffsetResetStrategy().name().toLowerCase(),
        true);

    // If the source is bounded, do not run periodic partition discovery.
    maybeOverride(
        KafkaSourceOptions.PARTITION_DISCOVERY_INTERVAL_MS.key(),
        "-1",
        boundedness == Boundedness.BOUNDED);

    // If the client id prefix is not set, reuse the consumer group id as the client id prefix.
    maybeOverride(
        KafkaSourceOptions.CLIENT_ID_PREFIX.key(),
        kafkaProperties.getProperty(ConsumerConfig.GROUP_ID_CONFIG),
        false);
  }
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