Spring boot auto-configure transaction id prefix

[nklmish]

As per docs, we need to provide transactionIdPrefix in order to enable transaction support. It will be nice if we can add a field like transactionIdPrefix in Producer and use it in DefaultKafkaProducerFactory to support properties based configuration like spring.kafka.producer.transaction-id-prefix in application.properties .

Current Situation:
ATM, we have to manually create a bean just to enable the transaction support. E.g. (the code below is in Kotlin, but we have to do the same thing in Java)

    @Bean
    fun kafkaProducerFactory(): ProducerFactory<*, *> {
        val config = mapOf<String, Any>(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to "org.apache.kafka.common.serialization.StringSerializer",
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to "org.apache.kafka.common.serialization.StringSerializer",
                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
                ProducerConfig.ACKS_CONFIG to "all",
                ProducerConfig.BATCH_SIZE_CONFIG to "4000",
                ProducerConfig.COMPRESSION_TYPE_CONFIG to "snappy"
        )
        val defaultKafkaProducerFactory = DefaultKafkaProducerFactory<Any, Any>(config)
        defaultKafkaProducerFactory.setTransactionIdPrefix("txn-ingestor") // only this line is relevant 
        return defaultKafkaProducerFactory
    }

I am using:
Spring-KafkaVersion 2.0.1.RELEASE
SpringBootVersion 2.0.0.M6

If it sounds good than I can submit a PR.

[garyrussell]

Seems reasonable, but this is a boot issue, not a spring-kafka issue so the PR would be over there.

However, that is not the complete story since the container factory needs a KafkaTransactionManager if you want the container to start the transaction and send the offset(s) to the transaction.

BTW, a simpler work around would be

@Bean
public ProducerFactory<?, ?> kafkaProducerFactory(KafkaProperties properties) {
   DefaultKafkaProducerFactory<Object, Object> pf = new DefaultKafkaProducerFactory<>(
				this.properties.buildProducerProperties());
    pf.setTransactionIdPrefix("txn-ingestor");
    return pf;
}

[nklmish]

Thanks

However, that is not the complete story since the container factory needs a KafkaTransactionManager

In this case we will also need to provide @ConditionalOnMissingBean or something like that which creates KafkaTransactionManager bean when the property is enabled. Below is the update code.

    @Bean
    fun transactionManager(producerFactory : ProducerFactory<*, *>) : KafkaTransactionManager<*, *> {
        return KafkaTransactionManager(producerFactory)
    }

    @Bean
    fun kafkaProducerFactory(properties : KafkaProperties): ProducerFactory<*, *> {
        val defaultKafkaProducerFactory = DefaultKafkaProducerFactory<Any, Any>(properties.buildProducerProperties())
        defaultKafkaProducerFactory.setTransactionIdPrefix("txn-ingestor") // transaction-id can be application name or something in that line
        return defaultKafkaProducerFactory
    }

    @Bean
    fun kafkaTemplate(producerFactory : ProducerFactory<*, *>) : KafkaTemplate<*, *> = KafkaTemplate(producerFactory)

Is there something more which we need to configure by default when enabling transactional support for kafka?

[artembilan]

Please, raise an appropriate GH issue in the Spring Boot project and you may link from there to here.

This is kinda an invalid issue in this Spring Kafka project.

Thanks