Spring Boot Kafka消费时数据库连接池与事务异常排查求助

我们有一个同时作为Kafka消费者和生产者的Spring Boot微服务，通过@KafkaListener消费消息，在同一个Kafka线程内向其他主题生产消息。服务配置了3个分区，并发数设为3，当大量消息涌入时，出现连接池和事务相关异常。

异常堆栈信息 #

堆栈跟踪1 #

HikariPool-1 - Connection oracle.jdbc.driver.T4CConnection@31232d8f marked as broken because of SQLSTATE(08006), ErrorCode(17002) java.sql.SQLRecoverableException: IO Error at oracle.jdbc.driver.T4CPreparedStatement.executeForRows(T4CPreparedStatement.java:1048) at oracle.jdbc.driver.OracleStatement.executeSQLStatement(OracleStatement.java:1507)

堆栈跟踪2 #

java.lang.IllegalStateException: No value for key [HikariDataSource (null)] bound to thread at org.springframework.transaction.support.TransactionSynchronizationManager.unbindResource(TransactionSynchronizationManager.java:198) at org.springframework.orm.jpa.JpaTransactionManager.doCleanupAfterCompletion(JpaTransactionManager.java:628) 

核心代码实现 #

Kafka监听器类 #

@Service
@Slf4j
@Data
public class ConsumerMessengerNotifyMS {

    @Autowired ObjectMapper objectMapper;
    @Autowired MessengerProcessor messengerProcessor;

    @KafkaListener(
            id= "messageReceiver",
            clientIdPrefix = "ConsumerMessengerNotifyMS",
            topics = "${path.to.topic}",
            groupId = "${kafka.consumer-groups[1]}",
            autoStartup = "${spring.kafka.consumer.auto-startup}",
            containerFactory = "myFactory",
            batch = "true",
            concurrency = "3")
    public void onMessage(List<ConsumerRecord<String, String>> records) {
        messengerProcessor.process(records);
}

消息处理组件 #

@Slf4j
@Component
public class MessengerProcessor {

    @Autowired ObjectMapper objectMapper;
    @Autowired DeliveryService communicationDeliveryService;
    @Autowired NotificationService notificationService;
    @Autowired KafkaTemplate<Integer, String> kafkaTemplate;

    public void process(List<ConsumerRecord<String, String>> records) {
       records.stream()
                    .filter(Objects::nonNull)
                    .map(
                           record -> {
                                try {
                                    return this.objectMapper.readValue(record.value(), NotificationMessage.class);
                                } catch (JsonProcessingException jpe) {
                                    log.error("Error while deserializing Kafka Record {} to NotificationMessage with key {}:: ", record, record.key());
                                    return null;
                                }
                            })
                    .filter(Objects::nonNull)
                    .map(communicationDeliveryService::processNotificationMessage)
                    .collect(Collectors.toList())
                    .forEach(dto -> {
                        try {
                            ProducerRecord<Integer, String> producerRecord = new ProducerRecord<>("path.to.topic2", dto.getKey(), objectMapper.writeValueAsString(dto));
                            kafkaTemplate.send(producerRecord);
                        } catch (JsonProcessingException e) {
                            throw new RuntimeException(e);
                        }
                      });
       
}

业务处理方法（带事务） #

@Override
@Transactional(transactionManager = "myTransactionManager")
public Dto processNotificationMessage(NotificationMessage notificationMessage) {
            Dto dto = new Dto();
            CommunicationStatus communicationStatus = CommunicationStatus.valueOf(notificationMessage.getMessageStatus());
            String notificationId = notificationMessage.getOrderId();
            if (SMS.equals(notificationMessage.getMessageType())) {

                smsInfoRepository.updateStatus(communicationStatus,notificationId);
                DeliveryOrderHolder bySmsInfo = deliveryOrderRepository.findBySmsInfo_NotificationId(notificationId);
                //DeliveryOrderHolder是一个包含三个属性的对象
                dto.setInvoice(bySmsInfo.getInvoiceNumber());
                dto.setBatchId(bySmsInfo.getBatchId().longValue());
                dto.setKey(bySmsInfo.getId());
                
                                            
            } else if (EMAIL.equals(notificationMessage.getMessageType())) {
                emailInfoRepository.updateStatus(communicationStatus,notificationId);
                DeliveryOrderHolder byEmailInfo = deliveryOrderRepository.findByEmailInfo_NotificationId(notificationId);
                dto.setInvoice(byEmailInfo.getInvoiceNumber());
                dto.setBatchId(byEmailInfo.getBatchId().longValue());
                dto.setKey(byEmailInfo.getId());
            }
        return dto;
 }

数据访问层查询示例 #

@Modifying
@Query("Update EmailInfo emailInfo set emailInfo.status = :newStatus where emailInfo.notificationId = :notifId")
 int updateStatus(@Param("newStatus") CommunicationStatus status, @Param("notifId") String notificationId);

测试用消息生产者 #

@Scheduled(fixedDelay = 60000L)
public void pushKafka(){

        Random rd = new Random();
        for (int i = 0; i<batchSizeEmail; i++){
            DtoMessage dtoMessage = new DtoMessage();
            dtoMessage.setMessageType(rd.nextBoolean() ? SMS : EMAIL );
            dtoMessage.setMessageStatus(rd.nextBoolean() ? FAILED : SUBMITED);
            dtoMessage.setOrderId(String.valueOf(i));
            dtoMessage.setReceivedDate(LocalDateTime.now());
            try {
                kafkaTemplate.send("topic1", i, objectMapper.writeValueAsString(dtoMessage));
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }

}

环境与配置信息 #

技术栈版本 #

• Spring Boot：2.7.15
• Spring Kafka：2.8.11
• kafka-clients：3.1.2
• 数据库：Oracle Database 12c Enterprise Edition Release 12.1.0.2.0 - 64bit Production

核心依赖 #

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-jpa</artifactId>
<!--            <exclusions>-->
<!--                <exclusion>-->
<!--                    <groupId>org.apache.tomcat</groupId>-->
<!--                    <artifactId>tomcat-jdbc</artifactId>-->
<!--                </exclusion>-->
<!--            </exclusions>-->
</dependency>

<dependency>
    <groupId>com.oracle.database.jdbc</groupId>
    <artifactId>ojdbc8</artifactId>
    <version>21.1.0.0</version>
</dependency>
<dependency>
    <groupId>com.oracle.database.jdbc</groupId>
    <artifactId>ucp</artifactId>
    <version>21.1.0.0</version>
</dependency>

数据源与事务配置类 #

@EnableAutoConfiguration
@Configuration
@EnableTransactionManagement
@EnableJpaRepositories(
        entityManagerFactoryRef = "myEntityManagerFactory",
        transactionManagerRef = "myTransactionManager",
        basePackages = {"root.package"})
public class ApplicationConfiguration {

    @Primary
    @Bean(name = "dataSource")
    public DataSource dataSource() throws SQLException {
       // OracleDataSource
//        PoolDataSource poolDataSource = PoolDataSourceFactory.getPoolDataSource();
//        poolDataSource.setConnectionFactoryClassName("oracle.jdbc.pool.OracleDataSource");
//        poolDataSource.setURL("url");
//        poolDataSource.setUser("user");
//        poolDataSource.setPassword("pass");
//        poolDataSource.setMaxStatements(10);
//        poolDataSource.setInitialPoolSize(10);
//        poolDataSource.setMaxPoolSize(30);
//        return poolDataSource;

         return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }

    @Primary
    @Bean("myEntityManagerFactory")
    public LocalContainerEntityManagerFactoryBean myEntityManagerFactory() throws SQLException {

        HibernateJpaVendorAdapter jpaVendorAdapter = new HibernateJpaVendorAdapter();
        jpaVendorAdapter.setGenerateDdl(false);
        jpaVendorAdapter.setDatabase(Database.ORACLE);
        LocalContainerEntityManagerFactoryBean factoryBean = new LocalContainerEntityManagerFactoryBean();
        factoryBean.setDataSource(dataSource());
        factoryBean.setJpaVendorAdapter(jpaVendorAdapter);
        factoryBean.setPackagesToScan(ClassUtils.getPackageName(AllDomainEntity.class));
      
        return factoryBean;
    }

    @Primary
    @Bean(name = "myTransactionManager")
    public JpaTransactionManager myTransactionManager(EntityManagerFactory emf) throws SQLException {
        return new JpaTransactionManager(emf);
    }

}

已尝试的排查方案 #

• 切换连接池：尝试使用Oracle UCP连接池替代HikariCP
• 更换事务管理器：测试过DataSourceTransactionManager、JtaTransactionManager
• 调整事务传播级别：将@Transactional的传播级别设为REQUIRES_NEW
• 事务范围调整：直接在Kafka监听器方法上添加@Transactional
• 数据访问方式变更：改用原生SQL查询、JdbcTemplate替代JPA

问题 #

异常出现时机不稳定，有时立即触发，有时在消费几分钟后出现，仅在测试环境复现。需要找到遗漏的配置或代码问题，确保所有Kafka消息都能被正常处理（不关心处理顺序）。

内容的提问来源于stack exchange，提问作者SOme1