Unable to send message when Observability Enabled

[hemantmaersk]

Description:
When observability is enabled in a Spring Boot application using Spring Kafka, we are encountering an issue where the application is unable to send message using kafkaTemplate. As per logs, it seems the spring code for observability is not obtain Kafka cluster information. This issue results in a TimeoutException during the clusterId call. Please note that we are able to send message with spring.kafka.template.observation-enabled: false but could not send message with spring.kafka.template.observation-enabled: true. The relevant portions of the stack trace are as follows:

Steps to Reproduce:

1. Enable observability setting spring.kafka.template.observation-enabled to true
2. Use Spring Kafka in a Spring Boot application version 3.2.0
3. Send message using kafkaTemplate
4. Application will not be able to send message using kafkaTemplate.
5. Now disable observability by setting spring.kafka.template.observation-enabled to false
6. Send message using kafkaTemplate
7. Application will be able to send message.

Expected Behavior:
The application should be able to send message by obtaining Kafka cluster information successfully even when observability is enabled.

Actual Behavior:
The TimeoutException occurs during the clusterId call, and the application fails to obtain Kafka cluster information.

Environment:

Spring Boot version: 3.2.0
Kafka version: 3.4.0

logs

16:14:29.750 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- [AdminClient clientId=adminclient-1] Node -1 disconnected.
16:14:29.750 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 43 due to node -1 being disconnected (elapsed time since creation: 212ms, elapsed time since send: 212ms, request timeout: 3600000ms)
16:14:29.869 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- [AdminClient clientId=adminclient-1] Metadata update failed
org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: fetchMetadata
16:14:29.971 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- App info kafka.admin.client for adminclient-1 unregistered
16:14:29.971 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- [AdminClient clientId=adminclient-1] Metadata update failed
org.apache.kafka.common.errors.TimeoutException: The AdminClient thread has exited. Call: fetchMetadata
16:14:29.971 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- [AdminClient clientId=adminclient-1] Timed out 1 remaining operation(s) during close.
16:14:29.973 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- Metrics scheduler closed
16:14:29.973 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- Closing reporter org.apache.kafka.common.metrics.JmxReporter
16:14:29.973 [kafka-admin-client-thread | adminclient-1] INFO  [peer-routing, traceId: , spanId: ] --- Metrics reporters closed
16:14:29.996 [http-nio-8085-exec-2] ERROR [peer-routing, traceId: 6571a1d1550947dbb1d77f8b3073c664, spanId: b1d77f8b3073c664] --- Could not obtain cluster info
java.util.concurrent.TimeoutException: null
    at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1960)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2095)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:180)
    at org.springframework.kafka.core.KafkaAdmin.clusterId(KafkaAdmin.java:335)
    at org.springframework.kafka.core.KafkaTemplate.clusterId(KafkaTemplate.java:504)
    at org.springframework.kafka.support.micrometer.KafkaRecordSenderContext.<init>(KafkaRecordSenderContext.java:45)
    at org.springframework.kafka.core.KafkaTemplate.lambda$observeSend$3(KafkaTemplate.java:752)
    at io.micrometer.observation.Observation.createNotStarted(Observation.java:172)
    at io.micrometer.observation.docs.ObservationDocumentation.observation(ObservationDocumentation.java:188)
    at org.springframework.kafka.core.KafkaTemplate.observeSend(KafkaTemplate.java:750)
    at org.springframework.kafka.core.KafkaTemplate.send(KafkaTemplate.java:565)
    at com.maersk.fbm.integration.controller.RoutingController.sendKafkaMessage(RoutingController.java:144)
    at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
    at java.base/java.lang.reflect.Method.invoke(Method.java:577)
    at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:254)
    at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:182)
    at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:118)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:917)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:829)
    at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
    at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1089)
    at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:979)
    at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1014)
    at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:903)
    at jakarta.servlet.http.HttpServlet.service(HttpServlet.java:564)
    at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:885)
    at jakarta.servlet.http.HttpServlet.service(HttpServlet.java:658)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:205)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:149)
    at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:51)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:174)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:149)
    at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:174)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:149)
    at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:174)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:149)
    at org.springframework.web.filter.ServerHttpObservationFilter.doFilterInternal(ServerHttpObservationFilter.java:109)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:174)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:149)
    at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:116)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:174)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:149)
    at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:167)
    at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:90)
    at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:482)
    at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:115)
    at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:93)
    at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:74)
    at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:340)
    at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:391)
    at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:63)
    at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:896)
    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1744)
    at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:52)
    at org.apache.tomcat.util.threads.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1191)
    at org.apache.tomcat.util.threads.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:659)
    at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
    at java.base/java.lang.Thread.run(Thread.java:833)

[artembilan]

Not clear why your Kafka doesn't respond for the clusterId request, but looks like the solution provided for this issue should help you:

	/**
	 * Set the cluster id. Use this to prevent attempting to fetch the cluster id
	 * from the broker, perhaps if the user does not have admin permissions.
	 * @param clusterId the clusterId to set
	 * @since 3.1
	 */
	public void setClusterId(String clusterId) {

This property is not exposed to Spring Boot auto-configuration, but you may do something like:

@Bean 
String adminConfigurer(KafkaAdmin kafkaAdmin) {
    kafkaAdmin.setClusterId(...);
    return null;
}

[hemantmaersk]

HI @artembilan , In our organisation we dont get a Kafka user with admin privileges, it only has read and write permissions for topic. Are you suggesting that to acquired cluster id and enable observability we need admin rights?

[hemantmaersk]

update: It worked by configuring admin properties under kafka.admin.properties in yml making them same as that of kafka.producer.properties

[himanshu1316]

@hemantmaersk added kafka.admin.properties in our repo but did not work for me. Should we configure a Bean as well for this?

[sobychacko]

@himanshu1316 Can you create a new discussion or issue and provide more context? This is an old discussion that is closed. Thanks!

[rajshreeee]

@hemantmaersk added kafka.admin.properties in our repo but did not work for me. Should we configure a Bean as well for this?

Well, the whole of it is actually spring.kafka.admin.properties.

Also, spring docs mention that it will try to retrieve the cluster id only if it is not set manually. So, if you lack the permissions, go for it

    @Bean
    public KafkaAdmin kafkaAdmin() {
        var admin = new KafkaAdmin(
            Collections.singletonMap(
                AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
        );
        admin.setClusterId("default-cluster");
        return admin;
    }
    ```