Гетерогенные транзакции на примере Kafka vs. RDBMS

Имел ввиду, что организация распределенных транзакций через Кафка имеет много болерплейта и сложностей (вы их разобрали)

Хочется что-то вроде гарантии exactly ones для событий + единая транзакция для бизнес моделей в БД. Наткнулся на интересную статью, которая решает ряд проблем за счёт хранения offsets-ов Kafka вне брокера прям в БД

https://medium.com/p/91042e81c095

Ну вот получается, что тут я больше ссылался именно на второй кейс. Что отправка сообщений в kafka можно делать без CDC, используя всюду ID идемпотентности.

//метод getId возвращает ID, не который установлен брокером,
//а который установлен в коде, когда сообщение отправлялось.
var consumedMessage = consumer.consume();

//допустим есть таблица consumed_message c колонками
//id - id строки в БД (ни на что не влияет)
//consumed_message_id: UNIQUE INDEX - id обрабатываемого сообщения

//и есть таблица produced_message с колонками
//id - id строки в БД (ни на что не влияет)
//consumed_message_id - id обрабатываемого сообщения
//produced_message_id - id отправляемого сообщения
//produced_message_body: JSON - тело отправляемого сообщения

var messageWasConsumed = 
  select count(*) > 0
  	from consumed_message
		where consumed_message_id = consumedMessage.getId();

//здесь сообщения которые необходимо отправить
List producedMessage = new ArrayList<>();
if (!messageWasConsumed) {
  TRANSACTION BEGIN;
  
  //здесь любая бизнесовая логика с сохранениями в БД
  //и например в рамках этой логики необходимо отправить 
  //несколько сообщений, поэтому здесь происходит наполнение
  //массива producedMessages
  //с генерированными в коде id - producedMessage.getId()
  //и с телом - producedMessage.getBody()
  
  insert into consumed_message
    (consumed_message_id)
    values (consumedMessageId);
  
  for(var producedMessage in producedMessages) {
  	insert into produced_message 
    	(consmed_message_id, produced_message_id, produced_message_body)
    	values (consmedMessage.getId(), producedMessage.getId(), producedMessage.getBody());
  }
  
  TRANSACTOIN COMMIT;
} else {
  producedMessages =
		select produced_message_id, produced_message_body
    	from produced_message
    	where consumed_message_id = consumedMessage.getId();
}
  
for(var producedMessage in producedMessages) {
  producer.produce(producedMessage);
}
consumer.ack(message);

Тут важно, что каждый консьюмер имеет точно такую же структуру. Разница только внутри блока с бизнес-логикой.

Кажется, что данная структура позволяет решить оба кейса. И прием сообщения из kafka с записью в БД (первый кейс), который у вас описан, и одновременно отправку в kafka и запись в БД (второй кейс) без использования CDC.