Целью настройки transactional.id для продюсера, в моем понимании, является защита от производителя зомби и заблаговременное прерывание его транзакций, чтобы избежать необходимости ожидания тайм-аута.
Что яЯ заинтересован в том, чтобы иметь возможность продолжить сделку.Например:
producer.beginTransaction();
producer.send(...);
// now the producer crashes
// start a new producer with same transactional.id
producer = new KafkaProducer(...);
// this aborts the unfinished transaction of the previous producer
producer.initTransactions();
Мне кажется, что это возможно, просто API для этого просто не существует.Есть что-нибудь, что мешает нам сделать это?
Зачем мне это нужно?Транзакция Kafka в моем случае является частью более крупной распределенной транзакции, которая потерпела неудачу на втором этапе.Координатор транзакций сохранил свое состояние и был перезапущен и знает, что некоторые участники могли зафиксировать, а некоторые нет, поэтому требуется, чтобы все участники завершили фиксацию из предыдущего запуска.