private static Properties combineTableAndUnderlyingLogstoreProperties(
      Map<String, String> tableProperties, Properties producerConfig) {
    Properties finalProp;
    Properties underlyingLogStoreProps =
        CompatibleFlinkPropertyUtil.fetchLogstorePrefixProperties(tableProperties);
    if (producerConfig == null) {
      finalProp = underlyingLogStoreProps;
    } else {
      underlyingLogStoreProps
          .stringPropertyNames()
          .forEach(k -> producerConfig.putIfAbsent(k, underlyingLogStoreProps.get(k)));
      finalProp = producerConfig;
    }

    String logStoreAddress =
        CompatibleFlinkPropertyUtil.propertyAsString(tableProperties, LOG_STORE_ADDRESS, null);

    String logType =
        CompatibleFlinkPropertyUtil.propertyAsString(
            tableProperties, LOG_STORE_TYPE, LOG_STORE_STORAGE_TYPE_DEFAULT);
    if (logType.equals(LOG_STORE_STORAGE_TYPE_KAFKA)) {
      finalProp.putIfAbsent(
          "key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
      finalProp.putIfAbsent(
          "value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
      finalProp.putIfAbsent(
          "key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
      finalProp.putIfAbsent(
          "value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");

      if (logStoreAddress != null) {
        finalProp.putIfAbsent(BOOTSTRAP_SERVERS_CONFIG, logStoreAddress);
      }

      Preconditions.checkArgument(
          finalProp.containsKey(BOOTSTRAP_SERVERS_CONFIG),
          String.format("%s should be set", LOG_STORE_ADDRESS));
    }

    return finalProp;
  }
Extract and combine the properties for underlying log store queue.

@param tableProperties mixed-format table properties
@param producerConfig can be set by java API
@return properties with tableProperties and producerConfig which has higher priority.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/MixedFormatUtils.java