private List<TestTemplateInvocationContext> processClusterTestInternal(ExtensionContext context, ClusterTest annot, ClusterTestDefaults defaults) {
        Type[] types = annot.types().length == 0 ? defaults.types() : annot.types();
        Map<String, String> serverProperties = Stream.concat(Arrays.stream(defaults.serverProperties()), Arrays.stream(annot.serverProperties()))
                .filter(e -> e.id() == -1)
                .collect(Collectors.toMap(ClusterConfigProperty::key, ClusterConfigProperty::value, (a, b) -> b));

        Map<Integer, Map<String, String>> perServerProperties = Stream.concat(Arrays.stream(defaults.serverProperties()), Arrays.stream(annot.serverProperties()))
                .filter(e -> e.id() != -1)
                .collect(Collectors.groupingBy(ClusterConfigProperty::id, Collectors.mapping(Function.identity(),
                        Collectors.toMap(ClusterConfigProperty::key, ClusterConfigProperty::value, (a, b) -> b))));

        Map<Features, Short> features = Arrays.stream(annot.features())
                .collect(Collectors.toMap(ClusterFeature::feature, ClusterFeature::version));

        ClusterConfig config = ClusterConfig.builder()
                .setTypes(new HashSet<>(Arrays.asList(types)))
                .setBrokers(annot.brokers() == 0 ? defaults.brokers() : annot.brokers())
                .setControllers(annot.controllers() == 0 ? defaults.controllers() : annot.controllers())
                .setDisksPerBroker(annot.disksPerBroker() == 0 ? defaults.disksPerBroker() : annot.disksPerBroker())
                .setAutoStart(annot.autoStart() == AutoStart.DEFAULT ? defaults.autoStart() : annot.autoStart() == AutoStart.YES)
                .setListenerName(annot.listener().trim().isEmpty() ? null : annot.listener())
                .setServerProperties(serverProperties)
                .setPerServerProperties(perServerProperties)
                .setSecurityProtocol(annot.securityProtocol())
                .setMetadataVersion(annot.metadataVersion())
                .setTags(Arrays.asList(annot.tags()))
                .setFeatures(features)
                .build();

        return Arrays.stream(types).map(type -> type.invocationContexts(context.getRequiredTestMethod().getName(), config))
                .collect(Collectors.toList());
    }
This class is a custom JUnit extension that will generate some number of test invocations depending on the processing
of a few custom annotations. These annotations are placed on so-called test template methods. Template methods look
like normal JUnit test methods, but instead of being invoked directly, they are used as templates for generating
multiple test invocations.

Test class that use this extension should use one of the following annotations on each template method:

<ul>
    <li>{@link ClusterTest}, define a single cluster configuration</li>
    <li>{@link ClusterTests}, provide multiple instances of @ClusterTest</li>
    <li>{@link ClusterTemplate}, define a static method that generates cluster configurations</li>
</ul>

Any combination of these annotations may be used on a given test template method. If no test invocations are
generated after processing the annotations, an error is thrown.

Depending on which annotations are used, and what values are given, different {@link ClusterConfig} will be
generated. Each ClusterConfig is used to create an underlying Kafka cluster that is used for the actual test
invocation.

For example:

<pre>
&#64;ExtendWith(value = Array(classOf[ClusterTestExtensions]))
class SomeIntegrationTest {
  &#64;ClusterTest(brokers = 1, controllers = 1, clusterType = ClusterType.Both)
  def someTest(): Unit = {
    assertTrue(condition)
  }
}
</pre>

will generate two invocations of "someTest" (since ClusterType.Both was given). For each invocation, the test class
SomeIntegrationTest will be instantiated, lifecycle methods (before/after) will be run, and "someTest" will be invoked.
core/src/test/java/kafka/test/junit/ClusterTestExtensions.java