为什么kafkatransactionmanager没有应用到这个spring云流kafka生产者?

ijxebb2r  于 2021-06-04  发布在  Kafka
关注(0)|答案(1)|浏览(406)

我已经配置了一个spring云流kafka应用程序来使用事务(github上提供了完整的源代码):

spring:
  application:
    name: message-relay-service
  cloud:
    stream:
      kafka:
        binder:
          transaction:
            transaction-id-prefix: message-relay-tx-
            producer:
              configuration:
                retries: 1
                acks: all
                key:
                  serializer: org.apache.kafka.common.serialization.StringSerializer

      bindings:
        output:
          destination: transfer
          contentType: application/*+avro
      schema-registry-client:
        endpoint: http://localhost:8081
      schema:
        avro:
          subjectNamingStrategy: org.springframework.cloud.stream.schema.avro.QualifiedSubjectNamingStrategy
  datasource:
    url: jdbc:h2:tcp://localhost:9090/mem:mydb
    driver-class-name: org.h2.Driver
    username: sa
    password:
  jpa:
    hibernate:
      ddl-auto: create
    database-platform: org.hibernate.dialect.H2Dialect

server:
  port: 8085

此应用程序有一个计划任务,定期检查要使用 @Scheduled 任务。此方法的注解为 @Transactional 主类定义 @EnableTransactionManagement .
但是,在调试代码时,我意识到kafkatransactionmanager没有被执行,也就是说,没有kafka事务。有什么问题吗?

@EnableTransactionManagement
@EnableBinding(Source::class)
@EnableScheduling
@SpringBootApplication
class MessageRelayServiceApplication

fun main(args: Array<String>) {
    runApplication<MessageRelayServiceApplication>(*args)
}

---

@Component
class MessageRelay(private val outboxService: OutboxService,
                   private val source: Source) {

    @Transactional
    @Scheduled(fixedDelay = 10000)
    fun checkOutbox() {
        val pending = outboxService.getPending()
        pending.forEach {
            val message = MessageBuilder.withPayload(it.payload)
                    .setHeader(KafkaHeaders.MESSAGE_KEY, it.messageKey)
                    .setHeader(MessageHeaders.CONTENT_TYPE, it.contentType)
                    .build()
            source.output().send(message)
            outboxService.markAsProcessed(it.id)
        }
    }

}
p8ekf7hl

p8ekf7hl1#

我不明白 @EnableTransactionManagementaccount-service ,仅在 message-relay-service .
在任何情况下,您的方案目前都不受支持;事务绑定器是为这样的处理器设计的:使用者启动事务,使用者线程上发送的任何记录都参与该事务,使用者向事务发送偏移量,然后提交事务。
它不是专为生产者绑定;请针对活页夹打开github问题,因为它应该得到支持。
我不知道为什么你没有看到一个交易开始,但是,即使是这样,问题是 @Transactional 将使用boot的自动配置ktm(和producer-factory),绑定使用不同的producer-factory(您配置的那个)。
即使交易正在进行中,制作人也不会参与其中。

相关问题