Reuse User Certs From Existing KafkaUser

[masterlemmi]

WE have a working strimzi kafka cluster which has ACL enabled.
We also have several users successfully publishing/consuming messages.

I want to run my application locally on my machine. I have enabled router listener so I can expose my broker externally.
Then I downloaded the same user certificates used by the same application in the cluster.
However I am running into this error: TransactionalIdAuthorizationException: Transactional Id authorization failed.
Shouldn't I be able to reuse the same certificates?

[scholzj]

I think you would need to share the full logs and configurations in order for anyone to be able to help you. You should maybe also elaborate on the scenario -> are you trying to connect to two different Kafka clusters with the same user certificate? Or just to the same Kafka cluster - once from inside Kubernetes and once from outside Kubernetes?

[masterlemmi]

I'm just trying to run the application on my IDE while connected to the same kafka cluster. That's it.
I did create a new KafkaUser and downloaded the created keystore for my IDE's reference. But I'm getting the same error:

Here's the stack trace. MY googling indicates it has something to do with ACL. but I have the same ACL as the KafkaUser that can successfully consume messages.

1:41.682 [scheduling-1] ERROR o.s.s.s.TaskUtils$LoggingErrorHandler -- Unexpected error occurred in scheduled task
org.springframework.kafka.KafkaException: initTransactions() failed
	at org.springframework.kafka.core.DefaultKafkaProducerFactory.doCreateTxProducer(DefaultKafkaProducerFactory.java:933)
	at org.springframework.kafka.core.DefaultKafkaProducerFactory.createTransactionalProducer(DefaultKafkaProducerFactory.java:867)
	at org.springframework.kafka.core.DefaultKafkaProducerFactory.doCreateProducer(DefaultKafkaProducerFactory.java:778)
	at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:768)
	at org.springframework.kafka.core.ProducerFactoryUtils.getTransactionalResourceHolder(ProducerFactoryUtils.java:96)
	at org.springframework.kafka.core.KafkaTemplate.getTheProducer(KafkaTemplate.java:953)
	at org.springframework.kafka.core.KafkaTemplate.getTheProducer(KafkaTemplate.java:930)
	at org.springframework.kafka.core.KafkaTemplate.flush(KafkaTemplate.java:700)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:355)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:768)
	at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:379)
	at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184)
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:768)
	at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:720)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at org.springframework.scheduling.support.ScheduledMethodRunnable.runInternal(ScheduledMethodRunnable.java:130)
	at org.springframework.scheduling.support.ScheduledMethodRunnable.lambda$run$2(ScheduledMethodRunnable.java:124)
	at io.micrometer.observation.Observation.observe(Observation.java:499)
	at org.springframework.scheduling.support.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:124)
	at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572)
	at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:358)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
	at java.base/java.lang.Thread.run(Thread.java:1583)
Caused by: org.apache.kafka.common.errors.TransactionalIdAuthorizationException: Transactional Id authorization failed.

[scholzj]

Ok, great. So, what exactly does the ACL error on the broker side say? It will give you much more details than the client error. Also, make sure you have TLS authentication enabled on the external listener as well as that could cause this kind of issue.

[masterlemmi]

Yes. It is enabled if you're referring to this:
kafka: listeners: - name: xternal port: 9094 type: route tls: true

I checked the broker logs via oc logs -f but didn't find anything helpful while the error is being thrown on the client. I might need to tweak logging settings which I have yet to figure out how. I will get back. Thanks for the hints.

[scholzj]

The YAML is badly formatted. But it does not enable any authentication. So that would be something to fix. The broker logs should show a deny message for ACL issues. Maybe you are looking at a wrong broker?