@Override
  public void open() throws Exception {
    callback =
        (metadata, exception) -> {
          if (exception != null && asyncException == null) {
            asyncException = exception;
          }
          acknowledgeMessage();
        };
    producer = createProducer();
    transactionalProducer = createTransactionalProducer();
    transactionalProducer.initTransactions();
    partitions = getPartitionsByTopic(topic, producer);
    LOG.info("HiddenKafkaPartition topic:{}, partitions:{}.", topic, partitions);
  }
Errors encountered in the async producer are stored here.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/hidden/kafka/HiddenKafkaProducer.java