diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
index b7b3b7513..409b403de 100644
--- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
+++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
@@ -35,7 +35,6 @@ import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
-import org.springframework.cloud.function.context.FunctionCatalog;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
@@ -45,6 +44,7 @@ import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.FunctionRegistration;
 import org.springframework.cloud.function.context.catalog.FunctionAroundWrapper;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
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 3c393ca56..2433cf3e6 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
@@ -308,6 +308,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			throw new IllegalStateException("No capable binding targets found.");
 		}
 
+
+		BindingServiceProperties bsp = this.getBindingServiceProperties();
 		Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination,
 				outputChannel, producerMessageHandler instanceof Lifecycle producerMessageHandlerWithLifecycle
 			? producerMessageHandlerWithLifecycle : null) {
@@ -322,6 +324,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				return false;
 			}
 
+			@Override
+			public String getBinderName() {
+				return bsp == null ? null : bsp.getBinder(destination);
+			}
+
+			@Override
+			public String getBinderType() {
+				return bsp == null ? null : bsp.getBinderType(this.getBinderName());
+			}
+
 			@Override
 			public void afterUnbind() {
 				try {
@@ -480,6 +492,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				consumerEndpointWithLifecycle.start();
 			}
 
+			BindingServiceProperties bsp = this.getBindingServiceProperties();
 			Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(name,
 					group, inputChannel, consumerEndpoint instanceof Lifecycle consumerEndpointWithLifecycle
 							? consumerEndpointWithLifecycle : null) {
@@ -494,6 +507,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 					return true;
 				}
 
+				@Override
+				public String getBinderName() {
+					return bsp == null ? null : bsp.getBinder(name);
+				}
+
+				@Override
+				public String getBinderType() {
+					return bsp == null ? null : bsp.getBinderType(this.getBinderName());
+				}
+
 				@Override
 				protected void afterUnbind() {
 					try {
@@ -567,6 +590,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		if (properties.isAutoStartup() && resources.getSource() instanceof Lifecycle sourceWithLifecycle) {
 			sourceWithLifecycle.start();
 		}
+
+		BindingServiceProperties bsp = this.getBindingServiceProperties();
 		Binding<PollableSource<MessageHandler>> binding = new DefaultBinding<PollableSource<MessageHandler>>(
 				name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle sourceWithLifecycle
 						? sourceWithLifecycle : null) {
@@ -581,6 +606,16 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				return true;
 			}
 
+			@Override
+			public String getBinderName() {
+				return bsp == null ? null : bsp.getBinder(name);
+			}
+
+			@Override
+			public String getBinderType() {
+				return bsp == null ? null : bsp.getBinderType(this.getBinderName());
+			}
+
 			@Override
 			public void afterUnbind() {
 				afterUnbindConsumer(destination, this.group, properties);
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 d9994f57b..1d678cbbb 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
@@ -48,6 +48,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void start() {
 	}
 
@@ -58,6 +59,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void stop() {
 	}
 
@@ -77,6 +79,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void pause() {
 		this.stop();
 	}
@@ -89,6 +92,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void resume() {
 		this.start();
 	}
@@ -96,6 +100,7 @@ public interface Binding<T> extends Pausable {
 	/**
 	 * @return 'true' if the target component represented by this instance is running.
 	 */
+	@Override
 	default boolean isRunning() {
 		return false;
 	}
@@ -118,6 +123,26 @@ public interface Binding<T> extends Pausable {
 		return null;
 	}
 
+	/**
+	 * Returns the name of the binder for this binding.
+	 * @return binder name
+	 *
+	 * @since 4.0.2
+	 */
+	default String getBinderName() {
+		return null;
+	}
+
+	/**
+	 * Returns the type of the binder for this binding.
+	 * @return binder name
+	 *
+	 * @since 4.0.2
+	 */
+	default String getBinderType() {
+		return null;
+	}
+
 	/**
 	 * Unbinds the target component represented by this instance and stops any active
 	 * components. Implementations must be idempotent. After this method is invoked, the
diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java
index a4698fabc..a7b6575d1 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java
@@ -237,6 +237,11 @@ public class BindingServiceProperties
 		}
 	}
 
+	public String getBinderType(String binderName) {
+		BinderProperties bp = this.binders.get(binderName);
+		return bp != null ? bp.getType() : this.bindings.keySet().iterator().next();
+	}
+
 	public String getBinder(String bindingName) {
 		return getBindingProperties(bindingName).getBinder();
 	}
diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
index f445d7105..a38788767 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
@@ -135,19 +135,23 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
 		this.streamBridgeFunctionCache = new HashMap<>();
 	}
 
+	@Override
 	public boolean send(String bindingName, Object data) {
 		BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(bindingName);
 		MimeType contentType = StringUtils.hasText(bindingProperties.getContentType()) ? MimeType.valueOf(bindingProperties.getContentType()) : MimeTypeUtils.APPLICATION_JSON;
 		return this.send(bindingName, data, contentType);
 	}
 
+	@Override
 	public boolean send(String bindingName, Object data, MimeType outputContentType) {
 		return this.send(bindingName, null, data, outputContentType);
 	}
+	@Override
 	public boolean send(String bindingName, @Nullable String binderName, Object data) {
 		return this.send(bindingName, binderName, data, MimeTypeUtils.APPLICATION_JSON);
 	}
 
+	@Override
 	@SuppressWarnings({ "unchecked"})
 	public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) {
 		if (!this.initialized) {