From 0ddff6db93211707f5fdfb2a15d48b9c51f35e0d Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@vmware.com>
Date: Tue, 7 Jul 2020 10:53:44 -0400
Subject: [PATCH] Add BeanNameAware to Producer/Consumer Factories

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/930

Re-enable Hoxton to be used with Spring Boot 2.2.x.
---
 .../kafka/core/DefaultKafkaConsumerFactory.java            | 7 ++++++-
 .../kafka/core/DefaultKafkaProducerFactory.java            | 6 +++++-
 2 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
index 6f5f7fd688..52127d814e 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java
@@ -28,6 +28,7 @@
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.util.StringUtils;
@@ -57,7 +58,7 @@
  * @author Artem Bilan
  * @author Chris Gilbert
  */
-public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V> {
+public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>, BeanNameAware {
 
 	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(DefaultKafkaConsumerFactory.class));
 
@@ -104,6 +105,10 @@ public DefaultKafkaConsumerFactory(Map<String, Object> configs,
 		this.valueDeserializerSupplier = valueDeserializerSupplier == null ? () -> null : valueDeserializerSupplier;
 	}
 
+	@Override
+	public void setBeanName(String name) {
+	}
+
 	public void setKeyDeserializer(@Nullable Deserializer<K> keyDeserializer) {
 		this.keyDeserializerSupplier = () -> keyDeserializer;
 	}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
index 265cea4131..059514ee1c 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java
@@ -50,6 +50,7 @@
 import org.apache.kafka.common.serialization.Serializer;
 
 import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -107,7 +108,7 @@
  * @author Chris Gilbert
  */
 public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>, ApplicationContextAware,
-		ApplicationListener<ContextStoppedEvent>, DisposableBean {
+			BeanNameAware, ApplicationListener<ContextStoppedEvent>, DisposableBean {
 
 	/**
 	 * The default close timeout duration as 30 seconds.
@@ -208,6 +209,9 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.applicationContext = applicationContext;
 	}
+	@Override
+	public void setBeanName(String name) {
+	}
 
 	public void setKeySerializer(@Nullable Serializer<K> keySerializer) {
 		this.keySerializerSupplier = () -> keySerializer;