GH-1563: Fix JUnit5 @EmbeddedKafka.partitions()

Resolves #1563

Paritions property ignored in the JUnit5 condition.

Author: garyrussell

spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java

@@ -120,7 +120,8 @@ private EmbeddedKafkaBroker createBroker(EmbeddedKafka embedded) {
 		if (embedded.count() > 1 && ports.length == 1 && ports[0] == 0) {
 			ports = new int[embedded.count()];
 		}
-		broker = new EmbeddedKafkaBroker(embedded.count(), embedded.controlledShutdown(), embedded.topics())
+		broker = new EmbeddedKafkaBroker(embedded.count(), embedded.controlledShutdown(),
+						embedded.partitions(), embedded.topics())
 				.zkPort(embedded.zookeeperPort())
 				.kafkaPorts(ports)
 				.zkConnectionTimeout(embedded.zkConnectionTimeout())

spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/EmbeddedKafkaConditionTests.java

@@ -29,13 +29,15 @@
  * @since 2.3
  *
  */
-@EmbeddedKafka(bootstrapServersProperty = "my.bss.property", count = 2)
+@EmbeddedKafka(bootstrapServersProperty = "my.bss.property", count = 2, controlledShutdown = true, partitions = 3)
 public class EmbeddedKafkaConditionTests {
 
 	@Test
 	public void test(EmbeddedKafkaBroker broker) {
 		assertThat(broker.getBrokersAsString()).isNotNull();
 		assertThat(KafkaTestUtils.getPropertyValue(broker, "brokerListProperty")).isEqualTo("my.bss.property");
+		assertThat(KafkaTestUtils.getPropertyValue(broker, "controlledShutdown")).isEqualTo(Boolean.TRUE);
+		assertThat(broker.getPartitionsPerTopic()).isEqualTo(3);
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateIntegrationTests.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.
@@ -72,7 +72,7 @@
  *
  * @since 2.3.0
  */
-@EmbeddedKafka(topics = ReactiveKafkaProducerTemplateIntegrationTests.REACTIVE_INT_KEY_TOPIC, partitions = 1)
+@EmbeddedKafka(topics = ReactiveKafkaProducerTemplateIntegrationTests.REACTIVE_INT_KEY_TOPIC, partitions = 2)
 public class ReactiveKafkaProducerTemplateIntegrationTests {
 
 	private static final int DEFAULT_PARTITIONS_COUNT = 2;