TopicBuilder Improvements

- add `assignReplicas()`

* Increase test heap size

* Doc polishing - remove reference to embedded kafka in admin config

Author: garyrussell

build.gradle

@@ -138,7 +138,7 @@ subprojects { subproject ->
 	test {
 		// suppress all console output during testing unless running `gradle -i`
 		logging.captureStandardOutput(LogLevel.INFO)
-		maxHeapSize = '1024m'
+		maxHeapSize = '1536m'
 		jacoco {
 			append = false
 			destinationFile = file("$buildDir/jacoco.exec")

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

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.config;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -73,7 +74,22 @@ public TopicBuilder replicas(int replicaCount) {
 	 * @see NewTopic#replicasAssignments()
 	 */
 	public TopicBuilder replicasAssignments(Map<Integer, List<Integer>> replicaAssignments) {
-		this.replicasAssignments = replicaAssignments;
+		replicaAssignments.forEach((part, list) -> assignReplicas(part, list));
+		return this;
+	}
+
+	/**
+	 * Add an individual replica assignment.
+	 * @param partition the partition.
+	 * @param replicaList the replicas.
+	 * @return the builder.
+	 * @see NewTopic#replicasAssignments()
+	 */
+	public TopicBuilder assignReplicas(int partition, List<Integer> replicaList) {
+		if (this.replicasAssignments == null) {
+			this.replicasAssignments = new HashMap<>();
+		}
+		this.replicasAssignments.put(partition, new ArrayList<>(replicaList));
 		return this;
 	}
 

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

@@ -64,6 +64,9 @@ public class KafkaAdminTests {
 	@Autowired
 	private NewTopic topic2;
 
+	@Autowired
+	private NewTopic topic3;
+
 	@Test
 	public void testTopicConfigs() {
 		assertThat(topic1.configs()).containsEntry(
@@ -75,6 +78,7 @@ public void testTopicConfigs() {
 		assertThat(TopicBuilder.name("foo")
 					.replicas(3)
 					.build().replicationFactor()).isEqualTo((short) 3);
+		assertThat(topic3.replicasAssignments()).hasSize(3);
 	}
 
 	@Test
@@ -118,7 +122,7 @@ public static class Config {
 
 		@Bean
 		public EmbeddedKafkaBroker kafkaEmbedded() {
-			return new EmbeddedKafkaBroker(1);
+			return new EmbeddedKafkaBroker(3);
 		}
 
 		@Bean
@@ -146,6 +150,16 @@ public NewTopic topic2() {
 					.build();
 		}
 
+		@Bean
+		public NewTopic topic3() {
+			return TopicBuilder.name("baz")
+					.assignReplicas(0, Arrays.asList(0, 1))
+					.assignReplicas(1, Arrays.asList(1, 2))
+					.assignReplicas(2, Arrays.asList(2, 0))
+					.config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
+					.build();
+		}
+
 	}
 
 }

src/reference/asciidoc/kafka.adoc

@@ -18,8 +18,7 @@ The following example shows how to do so:
 @Bean
 public KafkaAdmin admin() {
     Map<String, Object> configs = new HashMap<>();
-    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
-            StringUtils.arrayToCommaDelimitedString(embeddedKafka().getBrokerAddresses()));
+    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
     return new KafkaAdmin(configs);
 }
 
@@ -40,6 +39,16 @@ public NewTopic topic2() {
             .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
             .build();
 }
+
+@Bean
+public NewTopic topic3() {
+    return TopicBuilder.name("thing3")
+            .assignReplicas(0, Arrays.asList(0, 1))
+            .assignReplicas(1, Arrays.asList(1, 2))
+            .assignReplicas(2, Arrays.asList(2, 0))
+            .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
+            .build();
+}
 ----
 ====