refactor(systemtest): Share KMS facades across RecordEncryptionST tests (kroxylicious#3703)

* perf(systemtest): Share KMS facades across RecordEncryptionST tests

Apply the shared KMS facade pattern from PR kroxylicious#3681 and kroxylicious#3700 to
RecordEncryptionST. This starts each KMS facade once and shares it
across all tests in the class, rather than starting/stopping for
each test method.

Changes:
- Replace @testtemplate + TestKubeKmsFacadeInvocationContextProvider
  with @ParameterizedClass and @MethodSource
- Add facadesSource() method using lazy stream evaluation with .peek()
- Use autoCloseArguments = true for automatic facade cleanup
- Replace @testtemplate + CompressionTypeInvocationContextProvider
  with @ParameterizedTest + @EnumSource(CompressionType.class) for
  compression testing

This approach allows three different parameter injection mechanisms
to work together:
- testKmsFacade from @parameter field (via @ParameterizedClass)
- namespace from KroxyliciousExtension (via AbstractSystemTests)
- compressionType from @EnumSource (via @ParameterizedTest)

Expected performance improvement: ~60% faster overall execution based
on similar changes to KmsIT (kroxylicious#3700) which showed 60% overall improvement
and 96.4% faster test execution for container-based KMS providers.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
Signed-off-by: Keith Wall <kwall@apache.org>

* reorder to allow parameter resolution to work.

Signed-off-by: Keith Wall <kwall@apache.org>

---------

Signed-off-by: Keith Wall <kwall@apache.org>
Co-authored-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: k-wall

kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/installation/kms/aws/LocalStack.java

@@ -11,6 +11,7 @@
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
 import java.nio.file.Path;
+import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
@@ -78,7 +79,9 @@ public void deploy() {
     // Experimental workaround for https://github.com/kroxylicious/kroxylicious/issues/3362
     @SuppressWarnings("java:S2095") // try-with-resources for "HttpClient" is not available until we adopt JDK21
     private void pingLocalStackEndpoint() {
-        var client = HttpClient.newHttpClient();
+        var client = HttpClient.newBuilder()
+                .connectTimeout(Duration.ofSeconds(5))
+                .build();
         Awaitility.await()
                 .atMost(30, TimeUnit.SECONDS)
                 .ignoreExceptions()

kroxylicious-systemtests/src/test/java/io/kroxylicious/systemtests/extensions/CompressionTypeInvocationContextProvider.java

@@ -9,27 +9,31 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Stream;
 
 import org.apache.kafka.common.record.CompressionType;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.Parameter;
+import org.junit.jupiter.params.ParameterizedClass;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.Pod;
 
 import io.kroxylicious.kms.service.TestKekManager;
 import io.kroxylicious.kms.service.TestKmsFacade;
+import io.kroxylicious.kms.service.TestKmsFacadeFactory;
 import io.kroxylicious.systemtests.AbstractSystemTests;
 import io.kroxylicious.systemtests.Constants;
 import io.kroxylicious.systemtests.clients.KafkaClients;
 import io.kroxylicious.systemtests.clients.records.ConsumerRecord;
-import io.kroxylicious.systemtests.extensions.CompressionTypeInvocationContextProvider;
-import io.kroxylicious.systemtests.extensions.TestKubeKmsFacadeInvocationContextProvider;
 import io.kroxylicious.systemtests.installation.kroxylicious.Kroxylicious;
 import io.kroxylicious.systemtests.installation.kroxylicious.KroxyliciousOperator;
 import io.kroxylicious.systemtests.k8s.exception.KubeClusterException;
@@ -44,6 +48,8 @@
 import static org.assertj.core.api.Assumptions.assumeThat;
 import static org.junit.jupiter.api.Assertions.assertAll;
 
+@ParameterizedClass(autoCloseArguments = true)
+@MethodSource("facadesSource")
 class RecordEncryptionST extends AbstractSystemTests {
     private static final Logger LOGGER = LoggerFactory.getLogger(RecordEncryptionST.class);
     private static final String MESSAGE = "Hello-world";
@@ -53,6 +59,19 @@ class RecordEncryptionST extends AbstractSystemTests {
     private TestKekManager testKekManager;
     private KroxyliciousOperator kroxyliciousOperator;
 
+    static Stream<? extends TestKmsFacade<?, ?, ?>> facadesSource() {
+        // We rely on the fact that streams are lazy so the facade isn't built
+        // or started until the first test needs it.
+        return TestKmsFacadeFactory.getTestKmsFacadeFactories()
+                .filter(f -> f.getClass().getName().contains("systemtests"))
+                .map(TestKmsFacadeFactory::build)
+                .filter(TestKmsFacade::isAvailable)
+                .peek(TestKmsFacade::start);
+    }
+
+    @Parameter
+    TestKmsFacade<?, ?, ?> testKmsFacade;
+
     @BeforeAll
     void setUp() {
         KafkaClients.getKafkaClient().preloadImage();
@@ -103,9 +122,8 @@ void afterEach(String namespace) {
         }
     }
 
-    @TestTemplate
-    @ExtendWith(TestKubeKmsFacadeInvocationContextProvider.class)
-    void ensureClusterHasEncryptedMessage(String namespace, TestKmsFacade<?, ?, ?> testKmsFacade) {
+    @Test
+    void ensureClusterHasEncryptedMessage(String namespace) {
         testKekManager = testKmsFacade.getTestKekManager();
         testKekManager.generateKek(KEK_PREFIX + topicName);
         int numberOfMessages = 1;
@@ -139,17 +157,12 @@ void ensureClusterHasEncryptedMessage(String namespace, TestKmsFacade<?, ?, ?> t
     /**
      * Produce and consume compressed message.
      *
-     * @param namespace the namespace
-     * @param testKmsFacade the test kms facade
      * @param compressionType the compression type
+     * @param namespace the namespace
      */
-    @TestTemplate
-    @ExtendWith(CompressionTypeInvocationContextProvider.class)
-    void produceAndConsumeCompressedMessages(String namespace, TestKmsFacade<?, ?, ?> testKmsFacade, CompressionType compressionType) {
-        produceAndConsumeMessage(namespace, compressionType, testKmsFacade);
-    }
-
-    private void produceAndConsumeMessage(String namespace, CompressionType compressionType, TestKmsFacade<?, ?, ?> testKmsFacade) {
+    @ParameterizedTest
+    @EnumSource(CompressionType.class)
+    void produceAndConsumeMessage(CompressionType compressionType, String namespace) {
         testKekManager = testKmsFacade.getTestKekManager();
         testKekManager.generateKek(KEK_PREFIX + topicName);
         int numberOfMessages = 1;
@@ -177,9 +190,8 @@ private void produceAndConsumeMessage(String namespace, CompressionType compress
     }
 
     @SuppressWarnings("java:S2925")
-    @TestTemplate
-    @ExtendWith(TestKubeKmsFacadeInvocationContextProvider.class)
-    void ensureClusterHasEncryptedMessageWithRotatedKEK(String namespace, TestKmsFacade<?, ?, ?> testKmsFacade) {
+    @Test
+    void ensureClusterHasEncryptedMessageWithRotatedKEK(String namespace) {
         // Skip AWS test execution because the ciphertext blob metadata to read the version of the KEK is not available anywhere
         assumeThat(isVaultKms(testKmsFacade)).isTrue();
         String kekAlias = KEK_PREFIX + topicName;
@@ -244,9 +256,8 @@ private void assertKekVersionWithinParcel(List<ConsumerRecord> consumerRecords,
     }
 
     @SuppressWarnings("java:S2925")
-    @TestTemplate
-    @ExtendWith(TestKubeKmsFacadeInvocationContextProvider.class)
-    void produceAndConsumeMessageWithRotatedKEK(String namespace, TestKmsFacade<?, ?, ?> testKmsFacade) {
+    @Test
+    void produceAndConsumeMessageWithRotatedKEK(String namespace) {
         String kekAlias = KEK_PREFIX + topicName;
         testKekManager = testKmsFacade.getTestKekManager();
         testKekManager.generateKek(kekAlias);