Topic Provisioning - default partitions/replicas

The `TopicBuilder` now supports deferring to the broker defaults
for number of partitions and replicas.

* Fix typo and check for negative partitions in KafkaAdmin to avoid bogus log.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/TopicBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,12 +20,14 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.common.config.TopicConfig;
 
 /**
- * Builder for a {@link NewTopic}.
+ * Builder for a {@link NewTopic}. Since 2.6 partitions and replicas default to
+ * {@link Optional#empty()} indicating the broker defaults will be applied.
  *
  * @author Gary Russell
  * @since 2.3
@@ -35,9 +37,9 @@ public final class TopicBuilder {
 
 	private final String name;
 
-	private int partitions = 1;
+	private Optional<Integer> partitions = Optional.empty();
 
-	private short replicas = 1;
+	private Optional<Short> replicas = Optional.empty();
 
 	private Map<Integer, List<Integer>> replicasAssignments;
 
@@ -48,22 +50,22 @@ private TopicBuilder(String name) {
 	}
 
 	/**
-	 * Set the number of partitions (default 1).
+	 * Set the number of partitions (default broker 'num.partitions').
 	 * @param partitionCount the partitions.
 	 * @return the builder.
 	 */
 	public TopicBuilder partitions(int partitionCount) {
-		this.partitions = partitionCount;
+		this.partitions = Optional.of(partitionCount);
 		return this;
 	}
 
 	/**
-	 * Set the number of replicas (default 1).
+	 * Set the number of replicas (default broker 'default.replication.factor').
 	 * @param replicaCount the replicas (which will be cast to short).
 	 * @return the builder.
 	 */
 	public TopicBuilder replicas(int replicaCount) {
-		this.replicas = (short) replicaCount;
+		this.replicas = Optional.of((short) replicaCount);
 		return this;
 	}
 

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

@@ -234,7 +234,7 @@ private Map<String, NewPartitions> checkPartitions(Map<String, NewTopic> topicNa
 			NewTopic topic = topicNameToTopic.get(n);
 			try {
 				TopicDescription topicDescription = f.get(this.operationTimeout, TimeUnit.SECONDS);
-				if (topic.numPartitions() < topicDescription.partitions().size()) {
+				if (topic.numPartitions() >= 0 && topic.numPartitions() < topicDescription.partitions().size()) {
 					LOGGER.info(() -> String.format(
 						"Topic '%s' exists but has a different partition count: %d not %d", n,
 						topicDescription.partitions().size(), topic.numPartitions()));

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

@@ -25,6 +25,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.clients.admin.AdminClient;
@@ -109,6 +110,32 @@ public void testAddTopicsAndAddPartitions() throws Exception {
 		adminClient.close(Duration.ofSeconds(10));
 	}
 
+	@Test
+	public void testDefaultPartsAndReplicas() throws Exception {
+		try (AdminClient adminClient = AdminClient.create(this.admin.getConfigurationProperties())) {
+			DescribeTopicsResult topics = adminClient.describeTopics(Arrays.asList("optBoth", "optPart", "optRepl"));
+			Map<String, TopicDescription> results = topics.all().get(10, TimeUnit.SECONDS);
+			var topicDescription = results.get("optBoth");
+			assertThat(topicDescription.partitions()).hasSize(2);
+			assertThat(topicDescription.partitions().stream()
+					.map(tpi -> tpi.replicas())
+					.flatMap(nodes -> nodes.stream())
+					.count()).isEqualTo(4);
+			topicDescription = results.get("optPart");
+			assertThat(topicDescription.partitions()).hasSize(2);
+			assertThat(topicDescription.partitions().stream()
+					.map(tpi -> tpi.replicas())
+					.flatMap(nodes -> nodes.stream())
+					.count()).isEqualTo(2);
+			topicDescription = results.get("optRepl");
+			assertThat(topicDescription.partitions()).hasSize(3);
+			assertThat(topicDescription.partitions().stream()
+					.map(tpi -> tpi.replicas())
+					.flatMap(nodes -> nodes.stream())
+					.count()).isEqualTo(6);
+		}
+	}
+
 	@Test
 	public void alreadyExists() throws Exception {
 		AtomicReference<Method> addTopics = new AtomicReference<>();
@@ -153,7 +180,8 @@ public static class Config {
 
 		@Bean
 		public EmbeddedKafkaBroker kafkaEmbedded() {
-			return new EmbeddedKafkaBroker(3);
+			return new EmbeddedKafkaBroker(3)
+					.brokerProperty("default.replication.factor", 2);
 		}
 
 		@Bean
@@ -192,6 +220,26 @@ public NewTopic topic3() {
 					.build();
 		}
 
+		@Bean
+		public NewTopic topic4() {
+			return TopicBuilder.name("optBoth")
+					.build();
+		}
+
+		@Bean
+		public NewTopic topic5() {
+			return TopicBuilder.name("optPart")
+					.replicas(1)
+					.build();
+		}
+
+		@Bean
+		public NewTopic topic6() {
+			return TopicBuilder.name("optRepl")
+					.partitions(3)
+					.build();
+		}
+
 	}
 
 }

src/reference/asciidoc/kafka.adoc

@@ -116,6 +116,34 @@ public NewTopic topic3() {
 ----
 ====
 
+Starting with version 2.6, you can omit `.partitions()` and/or `replicas()` and the broker defaults will be applied to those properties.
+The broker version must be at least 2.4.0 to support this feature - see https://cwiki.apache.org/confluence/display/KAFKA/KIP-464%3A+Defaults+for+AdminClient%23createTopic[KIP-464].
+
+====
+[source, java]
+----
+@Bean
+public NewTopic topic4() {
+    return TopicBuilder.name("defaultBoth")
+            .build();
+}
+
+@Bean
+public NewTopic topic5() {
+    return TopicBuilder.name("defaultPart")
+            .replicas(1)
+            .build();
+}
+
+@Bean
+public NewTopic topic6() {
+    return TopicBuilder.name("defaultRepl")
+            .partitions(3)
+            .build();
+}
+----
+====
+
 IMPORTANT: When using Spring Boot, a `KafkaAdmin` bean is automatically registered so you only need the `NewTopic` `@Bean` s.
 
 By default, if the broker is not available, a message is logged, but the context continues to load.