Блог

Я работаю над приложением, которое использует Kafka для получения сообщений из нескольких тем, сохраняя данные по мере их поступления.

С этой целью я использую класс @Service с несколькими методами, аннотированными @KafkaListener . Рассмотрим это:

 @Transactional
@KafkaListener(topics = MyFirstMessage.TOPIC, autoStartup = "false", containerFactory = "myFirstKafkaListenerContainerFactory")
public void handleMyFirstMessage(ConsumerRecord<String, MyFirstMessage> record, Acknowledgment acknowledgment) throws Exception {
    MyFirstMessage message = consume(record, acknowledgment);
    try {
        doHandle(record.key(), message);
    } catch (Exception e) {
        TransactionInterceptor.currentTransactionStatus().setRollbackOnly();
    } finally {
        acknowledgment.acknowledge();
    }
}

@Transactional
@KafkaListener(topics = MySecondMessage.TOPIC, autoStartup = "false", containerFactory = "mySecondKafkaListenerContainerFactory")
public void handleMySecondMessage(ConsumerRecord<String, MySecondMessage> record, Acknowledgment acknowledgment) throws Exception {
    MySecondMessage message = consume(record, acknowledgment);
    try {
        doHandle(record.key(), message);
    } catch (Exception e) {
        TransactionInterceptor.currentTransactionStatus().setRollbackOnly();
    } finally {
        acknowledgment.acknowledge();
    }
}
 

Пожалуйста, не обращайте внимания на информацию о setRollbackOnly, это не имеет отношения к этому вопросу.

Важно то, что методы doHandle() в каждом прослушивателе выполняют вставки в таблицу, которые иногда терпят неудачу, потому что автоматически сгенерированные ключи оказываются неуникальными после завершения окончательной фиксации.

Происходит то, что каждый метод doHandle() будет увеличивать ключевой столбец в своих собственных небольших транзакциях, и только один из них «выиграет» эту гонку. Другой завершится ошибкой во время фиксации с нарушением неуникального ограничения.

Как лучше всего справиться с этим? Как мне «синхронизировать» транзакции, чтобы они выполнялись как жемчужины в строке вместо всех сразу?

Я подумываю об использовании какого-то семафора или блокировки для сериализации вещей, но это пахнет как решение со многими подводными камнями. Если бы существовал общий шаблон или структура, помогающие решить эту проблему, мне было бы гораздо удобнее ее реализовать.

Смотрите документацию.

Использование @Transactional для DB и a KafkaTransactionManager в контейнере слушателя аналогично использованию a ChainedKafkaTransactionManager (настроенного с обоими TM) в контейнере. Передача данных в БД фиксируется, за ней следует Kafka, когда слушатель завершает работу в обычном режиме.

Когда слушатель выдает исключение, обе транзакции откатываются в том же порядке.

Это setRollbackOnly определенно относится к этому вопросу, поскольку при этом вы не откатываете транзакцию kafka.