From 169d56be2c100f63fade310c29665b0ef34a8b7b Mon Sep 17 00:00:00 2001
From: Soby Chacko <chackos@vmware.com>
Date: Wed, 15 Nov 2023 17:32:29 -0500
Subject: [PATCH] GH-2849: Pollable consumers bindings endpoint

 - With pollable consumers, bindings actuator endpoint throws
   an exception since it cannot find the binding key in the bindings
   map (the key is stroed with the actual destination topic).
   Addressing this issue by quering the binding with the binding name.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2849
---
 .../binder/AbstractMessageChannelBinder.java  | 43 ++++++++++++-------
 1 file changed, 27 insertions(+), 16 deletions(-)

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 867cd7dd2..f9daf5c0c 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
@@ -228,13 +228,37 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	}
 
 	private String resolveBinderName(String bindingName, BindingServiceProperties bindingServiceProperties) {
-		String binder = bindingServiceProperties == null ? null : bindingServiceProperties.getBindings().get(bindingName).getBinder();
+		String binder = resolveBinder(bindingName, bindingServiceProperties);
 		if (!StringUtils.hasText(binder)) {
 			return resolveFromDefaultBinder();
 		}
 		return binder;
 	}
 
+	private String resolveBinderType(String bindingName, BindingServiceProperties bindingServiceProperties) {
+		String binder = resolveBinder(bindingName, bindingServiceProperties);
+		if (!StringUtils.hasText(binder)) {
+			return resolveFromDefaultBinder();
+		}
+		else {
+			if (bindingServiceProperties.getBinders().get(binder) == null) {
+				return binder;
+			}
+			return bindingServiceProperties.getBinders().get(binder).getType();
+		}
+	}
+
+	private static String resolveBinder(String bindingName, BindingServiceProperties bindingServiceProperties) {
+		String binder = null;
+		if (bindingServiceProperties != null) {
+			BindingProperties bindingProperties = bindingServiceProperties.getBindings().get(bindingName);
+			if (bindingProperties != null) {
+				binder = bindingProperties.getBinder();
+			}
+		}
+		return binder;
+	}
+
 	private String resolveFromDefaultBinder() {
 		DefaultBinderTypeRegistry binderTypeRegistry =
 			AbstractMessageChannelBinder.this.getApplicationContext().getBean(DefaultBinderTypeRegistry.class);
@@ -246,19 +270,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return binderTypes.keySet().iterator().next();
 	}
 
-	private String resolveBinderType(String bindingName, BindingServiceProperties bindingServiceProperties) {
-		String binder = bindingServiceProperties == null ? null : bindingServiceProperties.getBindings().get(bindingName).getBinder();
-		if (!StringUtils.hasText(binder)) {
-			return resolveFromDefaultBinder();
-		}
-		else {
-			if (bindingServiceProperties.getBinders().get(binder) == null) {
-				return binder;
-			}
-			return bindingServiceProperties.getBinders().get(binder).getType();
-		}
-	}
-
 	/**
 	 * Binds an outbound channel to a given destination. The implementation delegates to
 	 * {@link ProvisioningProvider#provisionProducerDestination(String, ProducerProperties)}
@@ -640,12 +651,12 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 			@Override
 			public String getBinderName() {
-				return resolveBinderName(getBindingName(), bsp);
+				return resolveBinderName(properties.getBindingName(), bsp);
 			}
 
 			@Override
 			public String getBinderType() {
-				return resolveBinderType(getBindingName(), bsp);
+				return resolveBinderType(properties.getBindingName(), bsp);
 			}
 
 			@Override