Q:

Spring-Kafka with transactions, multiple threads

I got a case where the application consumes messages and produces messages as a response to the consumed messages.  This is done using kafka transactions, BUT the app also has a scheduled job that sends Kafka messages at regular intervals (also using transactions since it sends to two topics).  
When the scheduled job starts sending, I get this exception:
org.apache.kafka.common. KafkaException: TransactionalId aura-transaction-1: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION

Anyone know what might be the reason? 
I'm considering trying with different kafkaTemplates (+ producer factory) to see if that fixes the issue.  Since then I can assign a new transaction-id-prefix to the scheduled job.  Currently they have the same. 
Consumer uses a basic @KafkaListener that is already registered in a transaction from the KafkaMessageListenerContainer.  It then produces a message using KafkaTemplate.send(Object). 
The scheduled job uses the KafkaTemplate.executeInTransaction functionality and sends to two topics. 
Versions: 
Spring Boot 2.1.1
Spring Kafka: 2.2.2
StackTrace:
org.apache.kafka.common. KafkaException: TransactionalId person-identhendelse-lager-1.privat-person-fregIdenthendelse-v1.0: Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION
    at org.apache.kafka.clients.producer.internals. TransactionManager.transitionTo(TransactionManager.java:758)
    at org.apache.kafka.clients.producer.internals. TransactionManager.transitionTo(TransactionManager.java:751)
    at org.apache.kafka.clients.producer.internals. TransactionManager.beginTransaction(TransactionManager.java:216)
    at org.apache.kafka.clients.producer. KafkaProducer.beginTransaction(KafkaProducer.java:606)
    at org.springframework.kafka.core. DefaultKafkaProducerFactory$CloseSafeProducer.beginTransaction(DefaultKafkaProducerFactory.java:459)
    at org.springframework.kafka.core. KafkaTemplate.executeInTransaction(KafkaTemplate.java:278)
    at no.nav.person.identhendelse.lager.app.aggregat. AggregatIdenthendelsePublisher.sendForPerson(AggregatIdenthendelsePublisher.java:52)
    at no.nav.person.identhendelse.lager.app.aggregat. AggregatScheduledTask.aggregate(AggregatScheduledTask.java:54)
    at no.nav.person.identhendelse.lager.app.aggregat. AggregatScheduledTask$$FastClassBySpringCGLIB$$7f682c33.invoke(<generated>)
    at org.springframework.cglib.proxy. MethodProxy.invoke(MethodProxy.java:218)
    at org.springframework.aop.framework. CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:749)
    at org.springframework.aop.framework. ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
    at org.springframework.aop.aspectj. MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:88)
    at io.micrometer.core.aop. TimedAspect.timedMethod(TimedAspect.java:77)
    at sun.reflect. GeneratedMethodAccessor58.invoke(Unknown Source)
    at sun.reflect. DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect. Method.invoke(Method.java:498)
    at org.springframework.aop.aspectj. AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:644)
    at org.springframework.aop.aspectj. AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:633)
    at org.springframework.aop.aspectj. AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:70)
    at org.springframework.aop.framework. ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.transaction.interceptor. TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:294)
    at org.springframework.transaction.interceptor. TransactionInterceptor.invoke(TransactionInterceptor.java:98)
    at org.springframework.aop.framework. ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.aop.interceptor. ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:93)
    at org.springframework.aop.framework. ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at no.nav.person.utils.precondition.feature.annotation. PreconditionMethodInterceptor.invoke(PreconditionMethodInterceptor.java:22)
    at org.springframework.aop.framework. ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
    at org.springframework.aop.framework. CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
    at no.nav.person.identhendelse.lager.app.aggregat. AggregatScheduledTask$$EnhancerBySpringCGLIB$$e0b597f7.aggregate(<generated>)
    at sun.reflect. NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect. NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect. DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect. Method.invoke(Method.java:498)
    at org.springframework.scheduling.support. ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:84)
    at org.springframework.scheduling.support. DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at java.util.concurrent. Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent. FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent. ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent. ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    at java.util.concurrent. ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent. ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang. Thread.run(Thread.java:748)

Added example code:
https://github.com/Lg87/kafka-transaction-example
See readme.md and FIND KafkaException to see the exception that occurs. 

A:

When asking questions like this, always provide version information. 
Show your code and the complete stack trace. 
You mentioned transactionTemplate - don't use a template as well as executeInTransaction - they are redundant since they both start a transaction. 
We recently fixed a problem where such "nested" transactions were broken. 

EDIT
I found the problem; when using producerPerConsumerPartition (default true), producers used by the container should not be added to cache for use by arbitrary KafkaTemplate operations. 
As a work-around, use a different DefaultKafkaProducerFactory for the stand-alone template operations. 
https://github.com/spring-projects/spring-kafka/issues/908

