@Override
  public void sendToAllPartitions(LogData<T> logData) throws Exception {
    checkErroneous();
    byte[] message = logDataJsonSerialization.serialize(logData);
    List<ProducerRecord<byte[], byte[]>> recordList =
        IntStream.of(partitions)
            .mapToObj(i -> new ProducerRecord<byte[], byte[]>(topic, i, null, null, message))
            .collect(Collectors.toList());
    LOG.info("sending {} partitions with flip message={}.", recordList.size(), logData);
    long start = System.currentTimeMillis();
    try {
      transactionalProducer.beginTransaction();
      for (ProducerRecord<byte[], byte[]> producerRecord : recordList) {
        checkErroneous();
        transactionalProducer.send(producerRecord, callback);
      }
      transactionalProducer.commitTransaction();
      LOG.info("finished flips sending, cost {}ms.", System.currentTimeMillis() - start);
    } catch (Throwable e) {
      LOG.error("", e);
      transactionalProducer.abortTransaction();
      throw new FlinkRuntimeException(e);
    }
  }
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