Merge branch '3.4.x'

Closes gh-44438

Author: mhalbritter

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java

@@ -238,7 +238,7 @@ private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Bac
 
 	static void applySslBundle(Map<String, Object> properties, SslBundle sslBundle) {
 		if (sslBundle != null) {
-			properties.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, SslBundleSslEngineFactory.class.getName());
+			properties.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, SslBundleSslEngineFactory.class);
 			properties.put(SslBundle.class.getName(), sslBundle);
 		}
 	}

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java

@@ -216,8 +216,7 @@ public Configuration getConsumer() {
 			assertThat(context).hasSingleBean(KafkaConnectionDetails.class);
 			DefaultKafkaConsumerFactory<?, ?> consumerFactory = context.getBean(DefaultKafkaConsumerFactory.class);
 			Map<String, Object> configs = consumerFactory.getConfigurationProperties();
-			assertThat(configs).containsEntry("ssl.engine.factory.class",
-					"org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory");
+			assertThat(configs).containsEntry("ssl.engine.factory.class", SslBundleSslEngineFactory.class);
 			assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle);
 		});
 	}
@@ -310,8 +309,7 @@ public Configuration getProducer() {
 			assertThat(context).hasSingleBean(KafkaConnectionDetails.class);
 			DefaultKafkaProducerFactory<?, ?> producerFactory = context.getBean(DefaultKafkaProducerFactory.class);
 			Map<String, Object> configs = producerFactory.getConfigurationProperties();
-			assertThat(configs).containsEntry("ssl.engine.factory.class",
-					"org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory");
+			assertThat(configs).containsEntry("ssl.engine.factory.class", SslBundleSslEngineFactory.class);
 			assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle);
 		});
 	}
@@ -394,8 +392,7 @@ public Configuration getAdmin() {
 			assertThat(context).hasSingleBean(KafkaConnectionDetails.class);
 			KafkaAdmin admin = context.getBean(KafkaAdmin.class);
 			Map<String, Object> configs = admin.getConfigurationProperties();
-			assertThat(configs).containsEntry("ssl.engine.factory.class",
-					"org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory");
+			assertThat(configs).containsEntry("ssl.engine.factory.class", SslBundleSslEngineFactory.class);
 			assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle);
 		});
 	}
@@ -490,8 +487,7 @@ public Configuration getStreams() {
 					.getBean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME,
 							KafkaStreamsConfiguration.class)
 					.asProperties();
-				assertThat(configs).containsEntry("ssl.engine.factory.class",
-						"org.springframework.boot.autoconfigure.kafka.SslBundleSslEngineFactory");
+				assertThat(configs).containsEntry("ssl.engine.factory.class", SslBundleSslEngineFactory.class);
 				assertThat(configs).containsEntry("org.springframework.boot.ssl.SslBundle", sslBundle);
 			});
 	}