title: Why isn't the KafkaTransactionManager being applied to this Spring Cloud Stream Kafka Producer?
body: <p>I have configured a Spring Cloud Stream Kafka application to use transactions (<a href="https://github.com/codependent/kafka-outbox-pattern.git" rel="nofollow noreferrer">full source code available on Github</a>):</p>
  
  <pre><code>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
  </code></pre>
  
  <p>This app has a scheduled task that periodically checks records to send in a database using a <code>@Scheduled</code> task. This methods is annotated with <code>@Transactional</code> and the main class defines <code>@EnableTransactionManagement</code>.</p>
  
  <p>However when debugging the code I've realized that the KafkaTransactionManager isn't being executed, that is to say, there are no Kafka transactions in place. What's the problem?</p>
  
  <pre><code>@EnableTransactionManagement
  @EnableBinding(Source::class)
  @EnableScheduling
  @SpringBootApplication
  class MessageRelayServiceApplication
  
  fun main(args: Array&lt;String&gt;) {
      runApplication&lt;MessageRelayServiceApplication&gt;(*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)
          }
      }
  
  }
  
  </code></pre>