Kafka consumer fails after actuator bindings query and restart

Describe the issue

After querying the Kafka bindings actuator endpoint, stopping and starting a consumer binding causes the Kafka listener container to fail when it receives the next record. The record is not delivered to the application consumer.

The issue appears to be related to the workaround introduced for #3167. That workaround replaces the container's ObservationRegistry with a null implementation while serializing binding properties. The same container is subsequently restarted with that registry, and Micrometer fails when creating the Kafka observation because observationConfig() returns null.

A minimal reproducer is available here:
https://github.com/ferblaca/demoSCSKafka/tree/actuator-rebalance-issue

To Reproduce

1. Clone the reproducer and check out the actuator-rebalance-issue branch.

2. Start the Kafka and ZooKeeper services using the Docker Compose configuration.

3. Start the Spring Boot application.

4. Query the bindings actuator endpoint:

   curl http://localhost:8080/demo-stream-kafka/stream/bindings
   

5. Stop the consumer binding programmatically:

   curl -X POST http://localhost:8080/demo-stream-kafka/repro/stop
   

6. Start the consumer binding programmatically:

   curl -X POST http://localhost:8080/demo-stream-kafka/repro/start
   

7. Send a message to the Kafka destination:

   curl -X POST http://localhost:8080/demo-stream-kafka/repro/send
   

8. Observe the application log. The message is sent successfully, but the consumer callback is not invoked and the listener container reports the following error:

   java.lang.IllegalStateException: This error handler cannot process 'java.lang.NullPointerException's; no record information is available
   	at org.springframework.kafka.listener.DefaultErrorHandler.handleOtherException(DefaultErrorHandler.java:198) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:2139) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1436) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1825) ~[na:na]
   	at java.base/java.lang.Thread.run(Thread.java:1474) ~[na:na]
   Caused by: java.lang.NullPointerException: Cannot invoke \"io.micrometer.observation.ObservationRegistry$ObservationConfig.getObservationConvention(io.micrometer.observation.Observation$Context, io.micrometer.observation.ObservationConvention)\" because the return value of \"io.micrometer.observation.ObservationRegistry.observationConfig()\" is null
   	at io.micrometer.observation.Observation.createNotStarted(Observation.java:248) ~[micrometer-observation-1.17.1.jar:1.17.1]
   	at io.micrometer.observation.docs.ObservationDocumentation.observation(ObservationDocumentation.java:186) ~[micrometer-observation-1.17.1.jar:1.17.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2977) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:2806) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2692) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2323) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1621) ~[spring-kafka-4.1.1.jar:4.1.1]
   	at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1523) ~[spring-kafka-4.1.1.jar:4.1.1]
   

Version of the framework

• Spring Boot 4.1.1
• Spring Cloud Stream 5.0.3
• Spring Kafka 4.1.1
• Micrometer Observation 1.17.1
• Java 25.0.2

Expected behavior

Querying the bindings actuator endpoint must not modify the observation configuration of an active Kafka consumer container. After stopping and starting the binding, the consumer should continue to process records successfully without an exception in the listener container.

Screenshots

Not applicable.

Additional context

The reproducer indicates that the failure is triggered by this sequence:

1. Query bindings actuator endpoint.
2. Stop and start a Kafka consumer binding.
3. Publish a record.

Without querying the bindings endpoint, the consumer processes records normally. The suspected regression is related to the fix for #3167.

Start with the linked reproducer on the actuator-rebalance-issue branch and run its Docker Compose setup. Reproduce the sequence of querying /stream/bindings, stopping and restarting the consumer, then sending a record; inspect the Spring Cloud Stream workaround associated with #3167. Done means the restarted Kafka consumer processes the record without the observationConfig() failure.