Fix parameter check in Kafka cluster examples (#10361)

Author: winfriedgerlach

examples/kafka-cluster/src/test/java/com/example/kafkacluster/ApacheKafkaContainerCluster.java

@@ -25,12 +25,14 @@ public class ApacheKafkaContainerCluster implements Startable {
     private final Collection<KafkaContainer> brokers;
 
     public ApacheKafkaContainerCluster(String version, int brokersNum, int internalTopicsRf) {
-        if (brokersNum < 0) {
+        if (brokersNum <= 0) {
             throw new IllegalArgumentException("brokersNum '" + brokersNum + "' must be greater than 0");
         }
-        if (internalTopicsRf < 0 || internalTopicsRf > brokersNum) {
+        if (internalTopicsRf <= 0 || internalTopicsRf > brokersNum) {
             throw new IllegalArgumentException(
-                "internalTopicsRf '" + internalTopicsRf + "' must be less than brokersNum and greater than 0"
+                "internalTopicsRf '" +
+                internalTopicsRf +
+                "' must be less than or equal to brokersNum and greater than 0"
             );
         }
 
@@ -99,6 +101,6 @@ public void start() {
 
     @Override
     public void stop() {
-        this.brokers.stream().parallel().forEach(GenericContainer::stop);
+        this.brokers.parallelStream().forEach(GenericContainer::stop);
     }
 }

examples/kafka-cluster/src/test/java/com/example/kafkacluster/ConfluentKafkaContainerCluster.java

@@ -25,12 +25,14 @@ public class ConfluentKafkaContainerCluster implements Startable {
     private final Collection<ConfluentKafkaContainer> brokers;
 
     public ConfluentKafkaContainerCluster(String confluentPlatformVersion, int brokersNum, int internalTopicsRf) {
-        if (brokersNum < 0) {
+        if (brokersNum <= 0) {
             throw new IllegalArgumentException("brokersNum '" + brokersNum + "' must be greater than 0");
         }
-        if (internalTopicsRf < 0 || internalTopicsRf > brokersNum) {
+        if (internalTopicsRf <= 0 || internalTopicsRf > brokersNum) {
             throw new IllegalArgumentException(
-                "internalTopicsRf '" + internalTopicsRf + "' must be less than brokersNum and greater than 0"
+                "internalTopicsRf '" +
+                internalTopicsRf +
+                "' must be less than or equal to brokersNum and greater than 0"
             );
         }
 
@@ -100,6 +102,6 @@ public void start() {
 
     @Override
     public void stop() {
-        this.brokers.stream().parallel().forEach(GenericContainer::stop);
+        this.brokers.parallelStream().forEach(GenericContainer::stop);
     }
 }

examples/kafka-cluster/src/test/java/com/example/kafkacluster/KafkaContainerCluster.java

@@ -31,12 +31,14 @@ public class KafkaContainerCluster implements Startable {
     private final Collection<KafkaContainer> brokers;
 
     public KafkaContainerCluster(String confluentPlatformVersion, int brokersNum, int internalTopicsRf) {
-        if (brokersNum < 0) {
+        if (brokersNum <= 0) {
             throw new IllegalArgumentException("brokersNum '" + brokersNum + "' must be greater than 0");
         }
-        if (internalTopicsRf < 0 || internalTopicsRf > brokersNum) {
+        if (internalTopicsRf <= 0 || internalTopicsRf > brokersNum) {
             throw new IllegalArgumentException(
-                "internalTopicsRf '" + internalTopicsRf + "' must be less than brokersNum and greater than 0"
+                "internalTopicsRf '" +
+                internalTopicsRf +
+                "' must be less than or equal to brokersNum and greater than 0"
             );
         }
 

examples/kafka-cluster/src/test/java/com/example/kafkacluster/KafkaContainerKraftCluster.java

@@ -25,12 +25,14 @@ public class KafkaContainerKraftCluster implements Startable {
     private final Collection<KafkaContainer> brokers;
 
     public KafkaContainerKraftCluster(String confluentPlatformVersion, int brokersNum, int internalTopicsRf) {
-        if (brokersNum < 0) {
+        if (brokersNum <= 0) {
             throw new IllegalArgumentException("brokersNum '" + brokersNum + "' must be greater than 0");
         }
-        if (internalTopicsRf < 0 || internalTopicsRf > brokersNum) {
+        if (internalTopicsRf <= 0 || internalTopicsRf > brokersNum) {
             throw new IllegalArgumentException(
-                "internalTopicsRf '" + internalTopicsRf + "' must be less than brokersNum and greater than 0"
+                "internalTopicsRf '" +
+                internalTopicsRf +
+                "' must be less than or equal to brokersNum and greater than 0"
             );
         }
 
@@ -101,6 +103,6 @@ public void start() {
 
     @Override
     public void stop() {
-        this.brokers.stream().parallel().forEach(GenericContainer::stop);
+        this.brokers.parallelStream().forEach(GenericContainer::stop);
     }
 }