private void setupKafkaProperties() {
    if (tableProperties.containsKey(TableProperties.LOG_STORE_ADDRESS)) {
      kafkaProperties.put(
          BOOTSTRAP_SERVERS_CONFIG, tableProperties.get(TableProperties.LOG_STORE_ADDRESS));
    }
    if (tableProperties.containsKey(TableProperties.LOG_STORE_MESSAGE_TOPIC)) {
      setTopics(getLogTopic(tableProperties));
    }

    kafkaProperties.putIfAbsent(
        "properties.key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    kafkaProperties.putIfAbsent(
        "properties.value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    kafkaProperties.putIfAbsent(
        "properties.key.deserializer",
        "org.apache.kafka.common.serialization.ByteArrayDeserializer");
    kafkaProperties.putIfAbsent(
        "properties.value.deserializer",
        "org.apache.kafka.common.serialization.ByteArrayDeserializer");

    setupStartupMode();
  }
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