KafkaSingletons::enhanceConfig() causes ClassNotFoundException due to appending MetricReporter class name on empty string

[edzebo]

Is your feature request related to a problem? Please describe.
Hi all,
So, I have a bit of a specific issue with automatic instrumentation.
We are importing Kafka wrapper from a public library that sets "metric-reporters" by default to an empty string.
In our codebase, we are just setting up Consumer/Producer qualifiers, and the rest is taken care of automatically.
We can override that value for "public" so to say type of consumer/producer, but we cannot do that for the internal consumers/producers that are auto-inited by the library.

Hence on Producer/Consumer init, there is a ClassNotFoundException. (This works without instrumentation). Here is stacktrace:

org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:830)at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:666)
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:647)
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:627)
at com.*redacted*.start(ExampleConsumer.kt:171)
at com.*redacted*.consumer.ExampleConsumer.<init>(ExampleConsumer.kt:78)
at com.*redacted*.client.KafkaInit$Companion.buildConsumer(KafkaInit.kt:59)
...
...
...
at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:258)
at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:331)
at com.*redacted*example.Config$$SpringCGLIB$$0.getConsumer(<generated>)
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.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:139)
at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:655)
at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:493)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1332)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1162)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:560)
at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:520)
at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:326)
at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:234)
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:324)
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:200)
at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:973)
at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:917)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:584)
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:146)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:732)
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:434)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:310)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1304)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1293)
at com.*redacted*.ApplicationKt.main(Application.kt:24)
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.boot.loader.MainMethodRunner.run(MainMethodRunner.java:49)
at org.springframework.boot.loader.Launcher.launch(Launcher.java:95)
at org.springframework.boot.loader.Launcher.launch(Launcher.java:58)
at org.springframework.boot.loader.JarLauncher.main(JarLauncher.java:65)

Caused by: org.apache.kafka.common.KafkaException: Class  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.consumer.KafkaConsumer.buildMetrics(KafkaConsumer.java:879)
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:701)
... 49 common frames omitted

Caused by: java.lang.ClassNotFoundException: 
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:419)
at org.apache.kafka.common.utils.Utils.newInstance(Utils.java:408)
at org.apache.kafka.common.config.AbstractConfig.getConfiguredInstance(AbstractConfig.java:396)
... 53 common frames omitted

Due to this output line when in debug:
metric.reporters = [, io.opentelemetry.javaagent.shaded.instrumentation.kafka.internal.OpenTelemetryMetricsReporter]

And Caused by: org.apache.kafka.common.KafkaException: Class cannot be found (Note the double empty string) I managed to track it down to the ehanceConfig() function here:
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/instrumentation/kafka/kafka-clients/kafka-clients-0.11/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/kafkaclients/v0_11/KafkaSingletons.java#L87

This would append ,OpenTelemetryMetricsReporter class name and cause ClassNotFound downstream if metric.reporters was set to "" before.

Describe the solution you'd like
Something like this would work probably:

return (((String) class1).isBlank()) ? class2 : class1 + "," + class2;

Additional context
This is really niche case and I might be on the wrong path here as well, so any suggestions are more than welcome.

[edzebo]

Many thanks!