Factor User into a top-level repeatable annotation

api/src/main/java/io/kroxylicious/testing/kafka/api/KafkaClusterProvisioningStrategy.java

@@ -34,7 +34,7 @@ public interface KafkaClusterProvisioningStrategy {
      * @return The estimated provisioning time (including the time taken for {@link KafkaCluster#start()}.
      */
     Duration estimatedProvisioningTimeMs(List<Annotation> constraints,
-                                      Class<? extends KafkaCluster> declarationType);
+                                         Class<? extends KafkaCluster> declarationType);
 
     /**
      * Create a {@link KafkaCluster} instance with the given configuration.

impl/src/main/java/io/kroxylicious/testing/kafka/common/KafkaClusterConfig.java

@@ -86,6 +86,8 @@ public class KafkaClusterConfig {
             KRaftCluster.class,
             Tls.class,
             SaslPlainAuth.class,
+            User.class,
+            User.List.class,
             ZooKeeperCluster.class);
 
     public static boolean supportsConstraint(Class<? extends Annotation> annotation) {
@@ -116,10 +118,14 @@ public static KafkaClusterConfig fromConstraints(List<Annotation> annotations) {
             if (annotation instanceof SaslPlainAuth) {
                 builder.saslMechanism("PLAIN");
                 sasl = true;
-                builder.users(Arrays.stream(((SaslPlainAuth) annotation).value())
-                        .collect(Collectors.toMap(
-                                SaslPlainAuth.UserPassword::user,
-                                SaslPlainAuth.UserPassword::password)));
+            }
+            if (annotation instanceof User) {
+                builder.user(((User) annotation).user(), ((User) annotation).password());
+            }
+            if (annotation instanceof User.List) {
+                Arrays.stream(((User.List) annotation).value()).forEach(user -> {
+                    builder.user(user.user(), user.password());
+                });
             }
             if (annotation instanceof ClusterId) {
                 builder.kafkaKraftClusterId(((ClusterId) annotation).value());

impl/src/main/java/io/kroxylicious/testing/kafka/common/SaslPlainAuth.java

@@ -15,28 +15,12 @@
 
 /**
  * Annotation constraining a {@link KafkaClusterProvisioningStrategy} to use
- * provide a cluster that supports SASL-PLAIN configured with the
- * given users.
+ * provide a cluster that supports SASL-PLAIN. {@link User @User} annotations can be used to
+ * configure the cluster with users.
  */
 @Target({ ElementType.PARAMETER, ElementType.FIELD })
 @Retention(RetentionPolicy.RUNTIME)
 @KafkaClusterConstraint
 public @interface SaslPlainAuth {
 
-    /**
-     * @return The configured users, which must be non-empty.
-     */
-    UserPassword[] value();
-
-    @interface UserPassword {
-        /**
-         * @return A user name.
-         */
-        String user();
-
-        /**
-         * @return A password.
-         */
-        String password();
-    }
 }

impl/src/main/java/io/kroxylicious/testing/kafka/common/User.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.kroxylicious.testing.kafka.common;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Repeatable;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import io.kroxylicious.testing.kafka.api.KafkaClusterConstraint;
+
+@Target({ ElementType.PARAMETER, ElementType.FIELD })
+@Retention(RetentionPolicy.RUNTIME)
+@Repeatable(User.List.class)
+@KafkaClusterConstraint
+public @interface User {
+    /**
+     * @return A user name.
+     */
+    String user();
+
+    /**
+     * @return A password.
+     */
+    String password();
+
+    @Target({ ElementType.PARAMETER, ElementType.FIELD })
+    @Retention(RetentionPolicy.RUNTIME)
+    @KafkaClusterConstraint
+    @interface List {
+        User[] value();
+    }
+}

junit5-extension/src/test/java/io/kroxylicious/testing/kafka/junit5ext/ParameterExtensionTest.java

@@ -22,6 +22,7 @@
 import io.kroxylicious.testing.kafka.common.KRaftCluster;
 import io.kroxylicious.testing.kafka.common.SaslPlainAuth;
 import io.kroxylicious.testing.kafka.common.Tls;
+import io.kroxylicious.testing.kafka.common.User;
 import io.kroxylicious.testing.kafka.common.ZooKeeperCluster;
 import io.kroxylicious.testing.kafka.invm.InVMKafkaCluster;
 import kafka.server.KafkaConfig;
@@ -141,11 +142,25 @@ public void kraftBasedClusterParameter(@BrokerCluster @KRaftCluster KafkaCluster
     }
 
     @Test
-    public void saslPlainAuthenticatingClusterParameter(
-                                                        @BrokerCluster @SaslPlainAuth({
-                                                                @SaslPlainAuth.UserPassword(user = "alice", password = "foo"),
-                                                                @SaslPlainAuth.UserPassword(user = "bob", password = "bar")
-                                                        }) KafkaCluster cluster)
+    public void saslPlainAuthenticatingClusterParameterOneUser(
+                                                               @BrokerCluster @SaslPlainAuth @User(user = "alice", password = "foo") KafkaCluster cluster)
+            throws ExecutionException, InterruptedException {
+        var dc = describeCluster(cluster.getKafkaClientConfiguration("alice", "foo"));
+        assertEquals(1, dc.nodes().get().size());
+        assertEquals(cluster.getClusterId(), dc.clusterId().get());
+
+        var ee = assertThrows(ExecutionException.class, () -> describeCluster(cluster.getKafkaClientConfiguration("alice", "FOO")),
+                "Expect bad password to throw");
+        assertInstanceOf(SaslAuthenticationException.class, ee.getCause());
+
+        ee = assertThrows(ExecutionException.class, () -> describeCluster(cluster.getKafkaClientConfiguration("eve", "quux")),
+                "Expect unknown user to throw");
+        assertInstanceOf(SaslAuthenticationException.class, ee.getCause());
+    }
+
+    @Test
+    public void saslPlainAuthenticatingClusterParameterTwoUsers(
+                                                                @BrokerCluster @SaslPlainAuth @User(user = "alice", password = "foo") @User(user = "bob", password = "bar") KafkaCluster cluster)
             throws ExecutionException, InterruptedException {
         var dc = describeCluster(cluster.getKafkaClientConfiguration("alice", "foo"));
         assertEquals(1, dc.nodes().get().size());