GH-3167 Fix deserialization of ConatinerProperties

Nullify ObservationRegistry
Make ExpressionSerializer top level class since jackson 3 has issues with inner classes

Resolves #3167

Author: olegz

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -151,6 +151,10 @@
 import org.springframework.util.backoff.ExponentialBackOff;
 import org.springframework.util.backoff.FixedBackOff;
 
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.Observation.Scope;
+import io.micrometer.observation.ObservationRegistry;
+
 /**
  * A {@link org.springframework.cloud.stream.binder.Binder} that uses Kafka as the
  * underlying middleware.
@@ -861,6 +865,31 @@ private void propagateContainerProperties(ContainerProperties containerPropertie
 		containerProperties.setMissingTopicsFatal(listener.isMissingTopicsFatal());
 		containerProperties.setStopImmediate(listener.isImmediateStop());
 	}
+	
+	public static class NullObservationRegistry implements ObservationRegistry {
+
+		@Override
+		public @org.jspecify.annotations.Nullable Observation getCurrentObservation() {
+			return null;
+		}
+
+		@Override
+		public @org.jspecify.annotations.Nullable Scope getCurrentObservationScope() {
+			return null;
+		}
+
+		@Override
+		public void setCurrentObservationScope(@org.jspecify.annotations.Nullable Scope current) {
+			
+		}
+
+		@Override
+		public ObservationConfig observationConfig() {
+			return null;
+		}
+
+		
+	}
 
 	/**
 	 * Returns an unmodifiable copy of {@link ContainerProperties} associated with the destination name
@@ -872,6 +901,8 @@ private void propagateContainerProperties(ContainerProperties containerPropertie
 	@Override
 	protected Map<String, Object> doGetAdditionalConfigurationProperties(String destinationName) {
 		ContainerProperties kafkaContainerProperties = this.kafkaMessageListenerContainers.iterator().next().getContainerProperties();
+		// see 3167 we need to nullify ObservationRegistry to avoid jackson deserialization error
+		kafkaContainerProperties.setObservationRegistry(new NullObservationRegistry());
 		Map mapOfContainerProperties = this.objectMapper.convertValue(kafkaContainerProperties, Map.class);
 		Map<String, Object> additionalConfigurationProperties = new HashMap<>();
 		additionalConfigurationProperties.put("containerProperties", mapOfContainerProperties);

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExpressionSerializer.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2016-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.springframework.cloud.stream.binder;
+
+import org.springframework.expression.Expression;
+
+import tools.jackson.core.JacksonException;
+import tools.jackson.core.JsonGenerator;
+import tools.jackson.databind.SerializationContext;
+import tools.jackson.databind.ser.std.StdSerializer;
+
+/**
+ * since 5.0.2
+ * author Oleg Zhurakousky
+ */
+public class ExpressionSerializer extends StdSerializer<Expression> {
+
+	public ExpressionSerializer() {
+		super(Expression.class);
+	}
+	public ExpressionSerializer(Class<Expression> t) {
+		super(t);
+	}
+	@Override
+	public void serialize(Expression expression, JsonGenerator jsonGenerator, SerializationContext provider) throws JacksonException {
+		if (expression != null) {
+			jsonGenerator.writeString(expression.getExpressionString());
+		}
+	}
+}

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ExtendedProducerProperties.java

@@ -24,6 +24,10 @@ public class ExtendedProducerProperties<T> extends ProducerProperties {
 
 	private T extension;
 
+	public ExtendedProducerProperties() {
+
+	}
+
 	public ExtendedProducerProperties(T extension) {
 		this.extension = extension;
 	}
@@ -32,4 +36,7 @@ public T getExtension() {
 		return this.extension;
 	}
 
+	public void setExtension(T extension) {
+		this.extension = extension;
+	}
 }

core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java

@@ -215,6 +215,10 @@ public void setDynamicPartitionUpdatesEnabled(boolean enabled) {
 		this.dynamicPartitionUpdatesEnabled = enabled;
 	}
 
+	/**
+	 * @deprecated since 5.0.2 in favor of top level class in the same package
+	 */
+	@Deprecated
 	public static class ExpressionSerializer extends StdSerializer<Expression> {
 		public ExpressionSerializer() {
 			super(Expression.class);