Kafka ProducerFencedException при использовании транзакций с Spring Kafka и синхронизацией базы данных

Я сталкиваюсь с исключением ProducerFencedException при попытке синхронизировать события между Kafka и базой данных PostgreSQL, используя Spring Kafka. События отправляются в Kafka, и база данных обновляется успешно, и я могу видеть события в теме Kafka, когда использую уровень изоляции read_uncommitted. Однако потребитель, пох parece, не потребляет сообщения с уровнем изоляции read_committed.

Конфигурация продюсера:

@Configuration
public class KafkaProducerConfig {
    @Value("${bootstrap.servers}")
    private String boostrapServers;

    public Map<String, Object> producerConfig() {
        Map<String, Object> producerConfig = new HashMap<>();
        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, boostrapServers);
        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction-event-id");
        return producerConfig;
    }

    @Bean
    public <T> ProducerFactory<String, T> producerFactory() {
        DefaultKafkaProducerFactory<String, T> defaultKafkaProducerFactory =
                new DefaultKafkaProducerFactory<>(producerConfig());
        defaultKafkaProducerFactory.setTransactionIdPrefix("transaction-event-id");
        return defaultKafkaProducerFactory;
    }

    @Bean
    public <T> KafkaTemplate<String, T> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public KafkaTransactionManager<String, Object> transactionEventProducerKafkaTransactionManager() {
        return new KafkaTransactionManager<>(producerFactory());
    }
}

Конфигурация потребителя:

isolation.level=read_committed

Логика отправки события продюсером и обновления базы данных:

@Transactional
public void sendMessageToTopics(Map<Long, String> duplicateCampaignId,
                                List<CampaignSMS> campaignSMSList, SmsType smsType) {

    if (smsType == SmsType.BULK) {
        campaignSMSList.forEach(campaignSMS -> {
            sendEvent(smsBulkTopicName, campaignSMS);
        });
    }
    if (smsType == SmsType.PROFILE) {
        campaignSMSList.forEach(campaignSMS -> sendEvent(smsProfileTopicName, campaignSMS));
    }

    duplicateCampaignId.forEach((campaignId, errorMessage) -> customerCampaignRepository.
            updateCampaignSmsStatusById(campaignId, CustomerCampaignStatus.ERROR, errorMessage));
    customerCampaignRepository.batchUpdateCampaignSms(campaignSMSList);
}

private void sendEvent(String topicName, CampaignSMS campaignSMS) {
    try {
        kafkaTemplate.send(topicName, campaignSMS).get(3L, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
        log.error(e.getMessage());
        Thread.currentThread().interrupt();
        throw new InternalException(e.getMessage());
    } catch (ExecutionException | TimeoutException e) {
        log.error(e.getMessage());
        throw new InternalException(e.getMessage());
    }
}

Ошибка, которую я получаю в логах:

ERROR o.s.k.c.DefaultKafkaProducerFactory - commitTransaction failed: CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@39ec10a3]
org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId which fences the current one. ERROR o.s.k.c.DefaultKafkaProducerFactory - commitTransaction failed: CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@39ec10a3]
org.apache.kafka.common.errors.ProducerFencedException: Producer with transactionalId 'transaction-event-id0' and ProducerIdAndEpoch(producerId=120332, epoch=0) has been fenced by another producer with the same transactionalIdTransactionSynchronization.afterCompletion threw exception
org.apache.kafka.common.errors.ProducerFencedException: Producer with transactionalId 'transaction-event-id0' and ProducerIdAndEpoch(producerId=120332, epoch=0) has been fenced by another producer with the same transactionalId
org.apache.kafka.common.errors.ProducerFencedException: There is a newer producer with the same transactionalId which fences the current one.

Эта ошибка возникает при попытке зафиксировать транзакцию в моей запланированной задаче. Похоже, что несколько продюсеров используют один и тот же transactionalId, что приводит к этой проблеме с ограждением.Я также пытался динамически генерировать transactional.id с помощью UUID.randomUUID() для каждого продюсера, но этот подход также не сработал, и ошибка появляется та же.(ProducerConfig.TRANSACTIONAL_ID_CONFIG, “transaction-event-id”+UUID.randomUUID() и defaultKafkaProducerFactory.setTransactionIdPrefix(“transaction-event-id”+UUID.randomUUID());)

Вот мои вопросы:

1. Как я могу гарантировать, что каждый экземпляр продюсера имеет уникальный
   transactionalId?
2. Каковы лучшие практики управления транзакциями Kafka при
   синхронизации с базой данных?
3. Существует ли способ обработать ProducerFencedException плавно в таком
   сценарии?

Любая помощь будет очень признательна! Спасибо!