GH-3483: Allow @OverRide of KafkaAdmin createAdmin()

Signed-off-by: Anders Swanson <[email protected]>

Author: anders-swanson

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

@@ -36,6 +36,7 @@
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.AlterConfigOp;
@@ -68,7 +69,7 @@
 import org.springframework.util.Assert;
 
 /**
- * An admin that delegates to an {@link AdminClient} to create topics defined
+ * An admin that delegates to an {@link Admin} to create topics defined
  * in the application context.
  *
  * @author Gary Russell
@@ -114,9 +115,9 @@ public class KafkaAdmin extends KafkaResourceFactory
 	private String clusterId;
 
 	/**
-	 * Create an instance with an {@link AdminClient} based on the supplied
+	 * Create an instance with an {@link Admin} based on the supplied
 	 * configuration.
-	 * @param config the configuration for the {@link AdminClient}.
+	 * @param config the configuration for the {@link Admin}.
 	 */
 	public KafkaAdmin(Map<String, Object> config) {
 		this.configs = new HashMap<>(config);
@@ -251,7 +252,7 @@ public void afterSingletonsInstantiated() {
 	public final boolean initialize() {
 		Collection<NewTopic> newTopics = newTopics();
 		if (!newTopics.isEmpty()) {
-			AdminClient adminClient = null;
+			Admin adminClient = null;
 			try {
 				adminClient = createAdmin();
 			}
@@ -347,7 +348,7 @@ protected Collection<NewTopic> newTopics() {
 	@Nullable
 	public String clusterId() {
 		if (this.clusterId == null) {
-			try (AdminClient client = createAdmin()) {
+			try (Admin client = createAdmin()) {
 				this.clusterId = client.describeCluster().clusterId().get(this.operationTimeout, TimeUnit.SECONDS);
 				if (this.clusterId == null) {
 					this.clusterId = "null";
@@ -365,14 +366,14 @@ public String clusterId() {
 
 	@Override
 	public void createOrModifyTopics(NewTopic... topics) {
-		try (AdminClient client = createAdmin()) {
+		try (Admin client = createAdmin()) {
 			addOrModifyTopicsIfNeeded(client, Arrays.asList(topics));
 		}
 	}
 
 	@Override
 	public Map<String, TopicDescription> describeTopics(String... topicNames) {
-		try (AdminClient admin = createAdmin()) {
+		try (Admin admin = createAdmin()) {
 			Map<String, TopicDescription> results = new HashMap<>();
 			DescribeTopicsResult topics = admin.describeTopics(Arrays.asList(topicNames));
 			try {
@@ -389,7 +390,7 @@ public Map<String, TopicDescription> describeTopics(String... topicNames) {
 		}
 	}
 
-	AdminClient createAdmin() {
+	protected Admin createAdmin() {
 		return AdminClient.create(getAdminConfig());
 	}
 
@@ -409,7 +410,7 @@ protected Map<String, Object> getAdminConfig() {
 		return configs2;
 	}
 
-	private void addOrModifyTopicsIfNeeded(AdminClient adminClient, Collection<NewTopic> topics) {
+	private void addOrModifyTopicsIfNeeded(Admin adminClient, Collection<NewTopic> topics) {
 		if (!topics.isEmpty()) {
 			Map<String, NewTopic> topicNameToTopic = new HashMap<>();
 			topics.forEach(t -> topicNameToTopic.compute(t.name(), (k, v) -> t));
@@ -439,7 +440,7 @@ private void addOrModifyTopicsIfNeeded(AdminClient adminClient, Collection<NewTo
 	}
 
 	private Map<ConfigResource, List<ConfigEntry>> checkTopicsForConfigMismatches(
-			AdminClient adminClient, Collection<NewTopic> topics) {
+			Admin adminClient, Collection<NewTopic> topics) {
 
 		List<ConfigResource> configResources = topics.stream()
 				.map(topic -> new ConfigResource(Type.TOPIC, topic.name()))
@@ -484,7 +485,7 @@ private Map<ConfigResource, List<ConfigEntry>> checkTopicsForConfigMismatches(
 		}
 	}
 
-	private void adjustConfigMismatches(AdminClient adminClient, Collection<NewTopic> topics,
+	private void adjustConfigMismatches(Admin adminClient, Collection<NewTopic> topics,
 			Map<ConfigResource, List<ConfigEntry>> mismatchingConfigs) {
 		for (Map.Entry<ConfigResource, List<ConfigEntry>> mismatchingConfigsOfTopic : mismatchingConfigs.entrySet()) {
 			ConfigResource topicConfigResource = mismatchingConfigsOfTopic.getKey();
@@ -556,7 +557,7 @@ else if (topic.numPartitions() > topicDescription.partitions().size()) {
 		return topicsToModify;
 	}
 
-	private void addTopics(AdminClient adminClient, List<NewTopic> topicsToAdd) {
+	private void addTopics(Admin adminClient, List<NewTopic> topicsToAdd) {
 		CreateTopicsResult topicResults = adminClient.createTopics(topicsToAdd);
 		try {
 			topicResults.all().get(this.operationTimeout, TimeUnit.SECONDS);
@@ -579,7 +580,7 @@ private void addTopics(AdminClient adminClient, List<NewTopic> topicsToAdd) {
 		}
 	}
 
-	private void createMissingPartitions(AdminClient adminClient, Map<String, NewPartitions> topicsToModify) {
+	private void createMissingPartitions(Admin adminClient, Map<String, NewPartitions> topicsToModify) {
 		CreatePartitionsResult partitionsResult = adminClient.createPartitions(topicsToModify);
 		try {
 			partitionsResult.all().get(this.operationTimeout, TimeUnit.SECONDS);

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.AlterConfigOp;
@@ -286,7 +287,7 @@ void nullClusterId() {
 		KafkaAdmin admin = new KafkaAdmin(Map.of()) {
 
 			@Override
-			AdminClient createAdmin() {
+			protected Admin createAdmin() {
 				return mock;
 			}