Add KafkaAdmin Auto Configuration

Spring for Apache Kafka has added a `KafkaAdmin` feature to automatically
add new topics to the broker.

See gh-10309

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

@@ -30,6 +30,7 @@ import org.springframework.context.annotation.Import;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaAdmin;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
@@ -103,4 +104,12 @@ public class KafkaAutoConfiguration {
 		return jaas;
 	}
 
+	@Bean
+	@ConditionalOnMissingBean(KafkaAdmin.class)
+	public KafkaAdmin kafkaAdmin() {
+		KafkaAdmin kafkaAdmin = new KafkaAdmin(this.properties.buildAdminProperties());
+		kafkaAdmin.setFatalIfBrokerNotAvailable(this.properties.getAdmin().isFailIfNoBrokers());
+		return kafkaAdmin;
+	}
+
 }

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

@@ -72,6 +72,8 @@ public class KafkaProperties {
 
 	private final Producer producer = new Producer();
 
+	private final Admin admin = new Admin();
+
 	private final Listener listener = new Listener();
 
 	private final Ssl ssl = new Ssl();
@@ -112,6 +114,10 @@ public class KafkaProperties {
 		return this.listener;
 	}
 
+	public Admin getAdmin() {
+		return this.admin;
+	}
+
 	public Ssl getSsl() {
 		return this.ssl;
 	}
@@ -186,6 +192,20 @@ public class KafkaProperties {
 		return properties;
 	}
 
+	/**
+	 * Create an initial map of admin properties from the state of this instance.
+	 * <p>
+	 * This allows you to add additional properties, if necessary, and override the
+	 * default kafkaAdmin bean.
+	 * @return the admin properties initialized with the customizations defined on this
+	 * instance
+	 */
+	public Map<String, Object> buildAdminProperties() {
+		Map<String, Object> properties = buildCommonProperties();
+		properties.putAll(this.admin.buildProperties());
+		return properties;
+	}
+
 	private static String resourceToPath(Resource resource) {
 		try {
 			return resource.getFile().getAbsolutePath();
@@ -643,6 +663,81 @@ public class KafkaProperties {
 
 	}
 
+	public static class Admin {
+
+		private final Ssl ssl = new Ssl();
+
+		/**
+		 * Id to pass to the server when making requests; used for server-side logging.
+		 */
+		private String clientId;
+
+		/**
+		 * Additional producer-specific properties used to configure the client.
+		 */
+		private final Map<String, String> properties = new HashMap<>();
+
+		/**
+		 * When true, the application context will not load if the broker connection
+		 * fails when attempting to provision topics.
+		 */
+		private boolean failIfNoBrokers;
+
+		public Ssl getSsl() {
+			return this.ssl;
+		}
+
+		public String getClientId() {
+			return this.clientId;
+		}
+
+		public void setClientId(String clientId) {
+			this.clientId = clientId;
+		}
+
+		public boolean isFailIfNoBrokers() {
+			return this.failIfNoBrokers;
+		}
+
+		public void setFailIfNoBrokers(boolean failIfNoBrokers) {
+			this.failIfNoBrokers = failIfNoBrokers;
+		}
+
+		public Map<String, String> getProperties() {
+			return this.properties;
+		}
+
+		public Map<String, Object> buildProperties() {
+			Map<String, Object> properties = new HashMap<>();
+			if (this.clientId != null) {
+				properties.put(ProducerConfig.CLIENT_ID_CONFIG, this.clientId);
+			}
+			if (this.ssl.getKeyPassword() != null) {
+				properties.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG,
+						this.ssl.getKeyPassword());
+			}
+			if (this.ssl.getKeystoreLocation() != null) {
+				properties.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG,
+						resourceToPath(this.ssl.getKeystoreLocation()));
+			}
+			if (this.ssl.getKeystorePassword() != null) {
+				properties.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,
+						this.ssl.getKeystorePassword());
+			}
+			if (this.ssl.getTruststoreLocation() != null) {
+				properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
+						resourceToPath(this.ssl.getTruststoreLocation()));
+			}
+			if (this.ssl.getTruststorePassword() != null) {
+				properties.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
+						this.ssl.getTruststorePassword());
+			}
+			properties.putAll(this.properties);
+			return properties;
+		}
+
+	}
+
 	public static class Template {
 
 		/**