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 044b0d698..b502a5e53 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
@@ -272,6 +272,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 						this.headersToEmbed, useNativeEncoding(producerProperties)));
 
 
+
+		BindingServiceProperties bsp = this.getBindingServiceProperties();
 		Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination,
 				outputChannel, producerMessageHandler instanceof Lifecycle
 						? (Lifecycle) producerMessageHandler : null) {
@@ -286,6 +288,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 {
@@ -440,6 +452,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				((Lifecycle) consumerEndpoint).start();
 			}
 
+			BindingServiceProperties bsp = this.getBindingServiceProperties();
 			Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(name,
 					group, inputChannel, consumerEndpoint instanceof Lifecycle
 							? (Lifecycle) consumerEndpoint : null) {
@@ -454,6 +467,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 {
@@ -527,6 +550,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		if (properties.isAutoStartup() && resources.getSource() instanceof Lifecycle) {
 			((Lifecycle) resources.getSource()).start();
 		}
+
+		BindingServiceProperties bsp = this.getBindingServiceProperties();
 		Binding<PollableSource<MessageHandler>> binding = new DefaultBinding<PollableSource<MessageHandler>>(
 				name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle
 						? (Lifecycle) resources.getSource() : null) {
@@ -541,6 +566,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 bef38c71f..d0b5c0ea9 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
@@ -49,6 +49,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void start() {
 	}
 
@@ -59,6 +60,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void stop() {
 	}
 
@@ -78,6 +80,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void pause() {
 		this.stop();
 	}
@@ -90,6 +93,7 @@ public interface Binding<T> extends Pausable {
 	 *
 	 * @see BindingsEndpoint
 	 */
+	@Override
 	default void resume() {
 		this.start();
 	}
@@ -97,6 +101,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;
 	}
@@ -119,6 +124,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();
 	}