org.apache.kafka.common.errors.TransactionAbortableException: 事务已被协调器中止

解决方案

1. Retry the transaction with a new transactional.id after aborting the current one.
   Code example:
   producer.initTransactions();
   try {
     producer.beginTransaction();
     // send messages
     producer.commitTransaction();
   } catch (TransactionAbortableException e) {
     producer.abortTransaction();
     // Reinitialize with a new transactional.id
     properties.put("transactional.id", "new-txn-id-" + System.currentTimeMillis());
     producer.close();
     producer = new KafkaProducer<>(properties);
     producer.initTransactions();
   }

2. Increase transaction coordinator timeout and ensure stable network.
   Command to check coordinator:
   kafka-transactions --bootstrap-server localhost:9092 --describe --transactional-id my-txn-id
   Set in producer config:
   properties.put("transaction.timeout.ms", "60000");
   properties.put("max.block.ms", "120000");