From adddeffbf33644fc6a0bbd41a0b30618347f4dab Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Tue, 28 Jan 2025 09:36:58 +0100
Subject: [PATCH] GH-3071 Add support for accessing producer/consumer
 properties

---
 .../binder/AbstractMessageChannelBinder.java    | 16 ++++++++++++++++
 .../cloud/stream/binder/Binding.java            | 11 ++++++++++-
 .../binding/BindingsLifecycleController.java    | 17 +++++++++++++++++
 3 files changed, 43 insertions(+), 1 deletion(-)

diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
index e1ece1cbc..18b4fe92b 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
@@ -368,6 +368,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				return doGetExtendedInfo(destination, producerProperties);
 			}
 
+			@SuppressWarnings({ "unchecked", "hiding" })
+			public <P> P getExtension() {
+				if (producerProperties instanceof ExtendedProducerProperties extendedProperties) {
+					return (P) extendedProperties.getExtension();
+				}
+				return null;
+			}
+
 			@Override
 			public boolean isInput() {
 				return false;
@@ -551,6 +559,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 					return doGetExtendedInfo(destination, properties);
 				}
 
+				@SuppressWarnings({ "unchecked", "hiding" })
+				public <P> P getExtension() {
+					if (properties instanceof ExtendedConsumerProperties extendedProperties) {
+						return (P) extendedProperties.getExtension();
+					}
+					return null;
+				}
+
 				@Override
 				public boolean isInput() {
 					return true;
diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
index 1d678cbbb..674422d39 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2025 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.
@@ -41,6 +41,15 @@ public interface Binding<T> extends Pausable {
 		return Collections.emptyMap();
 	}
 
+	/**
+	 * Will return producer or consumer properties of this binding.
+	 * @param <P> type of producer or consumer properties
+	 * @return producer or consumer properties of this binding
+	 */
+	default <P> P getExtension() {
+		return null;
+	}
+
 	/**
 	 * Starts the target component represented by this instance. NOTE: At the time the
 	 * instance is created the component is already started. This operation is typically
diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java
index 9530f746e..009323328 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java
@@ -31,6 +31,7 @@ import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.CollectionUtils;
 
 /**
  *
@@ -52,6 +53,7 @@ public class BindingsLifecycleController {
 	@SuppressWarnings("unchecked")
 	public BindingsLifecycleController(List<InputBindingLifecycle> inputBindingLifecycles,
 			List<OutputBindingLifecycle> outputBindingsLifecycles) {
+
 		Assert.notEmpty(inputBindingLifecycles,
 				"'inputBindingLifecycles' must not be null or empty");
 		this.inputBindingLifecycles = inputBindingLifecycles;
@@ -72,6 +74,21 @@ public class BindingsLifecycleController {
 		}
 	}
 
+	/**
+	 * Will return producer or consumer properties for a specified binding. For example, calling `getExtensionProperties("foo-in-0")`
+	 * on Kafka binding  will return an instance of KafkaConsumerProperties.
+	 * @param <T> type of producer or consumer properties for a specified binding
+	 * @param bindingName name of the binding
+	 * @return producer or consumer properties
+	 */
+	public <T> T getExtensionProperties(String bindingName) {
+		List<Binding<?>> locateBinding = BindingsLifecycleController.this.locateBinding(bindingName);
+		if (!CollectionUtils.isEmpty(locateBinding)) {
+			return locateBinding.get(0).getExtension();
+		}
+		return null;
+	}
+
 	/**
 	 * Provide an accessor for the custom ObjectMapper created by this controller.
 	 * @return {@link ObjectMapper}