Caused by: java.lang.ClassNotFoundException: io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.OpenTelemetryMetricsReporter

[SanyaSingh-Groww]

I am getting this intermittent issue :

2024-03-12 21:00:32.478 ERROR c.e.p.listner.SftpReverseFlowListner : Error in reverse flow, file: /Groww_Balance/ReportCAN/NEXTBILLION_NBT_CAN_120320240832_320617800.txt

org.apache.kafka.common.KafkaException: org.apache.kafka.common.KafkaException: Failed to construct kafka producer
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:499)
at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:480)
at java.base/java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:562)
at java.base/java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:591)
at java.base/java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:689)
at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:159)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:173)
at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
at com.elements.payoutsiciciservice.reverseflow.ReverseFlowTxnUtil.updateTxn(ReverseFlowTxnUtil.java:104)
at com.elements.payoutsiciciservice.listner.SftpReverseFlowListner.listen(SftpReverseFlowListner.java:86)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:568)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:169)
at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:119)
at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:56)
at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:366)
at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92)
at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:53)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:2924)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:2904)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.lambda$doInvokeRecordListener$58(KafkaMessageListenerContainer.java:2822)
at io.micrometer.observation.Observation.lambda$observe$4(Observation.java:544)
at io.micrometer.observation.Observation.observeWithContext(Observation.java:603)
at io.micrometer.observation.Observation.observe(Observation.java:544)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2820)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:2672)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2558)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2200)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1555)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1519)
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1394)
at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1804)
at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.kafka.common.KafkaException: Failed to construct kafka producer
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:465)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:290)
at org.springframework.kafka.core.DefaultKafkaProducerFactory.createRawProducer(DefaultKafkaProducerFactory.java:884)
at org.springframework.kafka.core.DefaultKafkaProducerFactory.createKafkaProducer(DefaultKafkaProducerFactory.java:777)
at org.springframework.kafka.core.DefaultKafkaProducerFactory.doCreateProducer(DefaultKafkaProducerFactory.java:747)
at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:727)
at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:721)
at org.springframework.kafka.core.KafkaTemplate.getTheProducer(KafkaTemplate.java:918)
at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:772)
at org.springframework.kafka.core.KafkaTemplate.observeSend(KafkaTemplate.java:754)
at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:544)
at com.elements.payoutsiciciservice.reverseflow.ReverseFlowTxnUtil.lambda$updateTxn$1(ReverseFlowTxnUtil.java:150)
at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.base/java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:992)
at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at java.base/java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:290)
at java.base/java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:754)
at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:373)
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1182)
at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1655)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1622)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:165)
Caused by: org.apache.kafka.common.KafkaException: Class io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.OpenTelemetryMetricsReporter cannot be found
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:398)
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:480)
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstances(AbstractConfig.java:461)
at org.apache.kafka.clients.CommonClientConfigs.metricsReporters(CommonClientConfigs.java:237)
at org.apache.kafka.clients.CommonClientConfigs.metricsReporters(CommonClientConfigs.java:233)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:367)
... 21 common frames omitted
Caused by: java.lang.ClassNotFoundException: io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.OpenTelemetryMetricsReporter
at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:641)
at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:188)
at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:520)
at java.base/java.lang.Class.forName0(Native Method)
at java.base/java.lang.Class.forName(Class.java:467)
at org.apache.kafka.common.utils.Utils.loadClass(Utils.java:422)
at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:411)
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:396)
... 26 common frames omitted

dependecies version :

<spring-cloud-gcp.version>3.5.2</spring-cloud-gcp.version>
<spring.boot.version>3.1.0</spring.boot.version>

org.springframework.integration
spring-integration-kafka

To me it looks like version compatibilty issue but not sure what. Please help

[steverao]

The error appear when you use Otel Java Agent? If yes, please tell us which version do you use and you had better provide us a simplest demo.

[SanyaSingh-Groww]

I am not directly using this dependecy, looks like its there with <spring.boot.version>3.1.0</spring.boot.version>
@steverao also this is intermittent issue. happening very rarely.

[SanyaSingh-Groww]

can you tell which dependecy to add for this class : io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.OpenTelemetryMetricsReporter

[steverao]

can you tell which dependecy to add for this class : io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.OpenTelemetryMetricsReporter

The dependency of opentelemetry-javaagent contains the class you mentioned above. I recommend you show us how you use OpenTelemetry, it may can help us to figure out what root cause it is and provide you relevant solution.

[laurit]

You can disable the kafka metrics instrumentation that causes this issue by adding -Dotel.instrumentation.kafka-clients-metrics.enabled=false to jvm arguments. My guess is that context class loader of the thread that creates the kafka producer can't load the kafka classes (OpenTelemetryMetricsReporter should also be in the same class loader as kafka classes). Probably would also work when context class loader was set to null as then kafka would use the class loader that loaded kafka classes instead of the context class loader to load OpenTelemetryMetricsReporter.

[laurit]

Hopefully fixed in https://github.com/open-telemetry/opentelemetry-java-instrumentation/releases/tag/v2.2.0