Require opt-in to allow testing configuration

Adds a concept of configuration features. We want to add some features to the
proxy to assist in testing that we want to be disallowed unless the user
has opted into them, telling the proxy that these features are unlocked.

By default the proxy has no features enabled and will log an error and
fail to start if any testing configurations are specified in the config
file.

To opt in to using test-only configuration the user must supply either:
1. a system prop `-DKROXYLICIOUS_UNLOCK_TEST_ONLY_CONFIGURATION=true`
2. an env var `KROXYLICIOUS_UNLOCK_TEST_ONLY_CONFIGURATION=true`
with the system-prop taking precedence.

Integration tests behave the same way,you have to enable the feature
explicitly in your test to use any test-only features.

Why:
We plan to add some opt-in configuration to enable overriding the maximum
version the proxy accepts for an ApiKey. This is test only functionality
and we want to ensure this isn't accidentally applied in production
without some kind of opt-in hoops to leap through.

Signed-off-by: Robert Young <robeyoun@redhat.com>

Author: robobario

CHANGELOG.md

@@ -7,6 +7,7 @@ Format `<github issue/pr number>: <short description>`.
 
 ## SNAPSHOT
 
+* [#1648](https://github.com/kroxylicious/kroxylicious/pull/1648) Add test-only feature mechanism to Proxy configuration
 * [#1379](https://github.com/kroxylicious/kroxylicious/issues/1379) Remove Deprecated EnvelopeEncryption
 * [#1561](https://github.com/kroxylicious/kroxylicious/pull/1631) Allow Trust and ClientAuth to be set for Downstream TLS
 * [#1550](https://github.com/kroxylicious/kroxylicious/pull/1550) Upgrade Apache Kafka from 3.8.0 to 3.9.0 #1550

kroxylicious-app/src/main/java/io/kroxylicious/app/Kroxylicious.java

@@ -8,9 +8,9 @@
 import java.io.File;
 import java.io.InputStream;
 import java.nio.file.Files;
+import java.util.Arrays;
 import java.util.Properties;
 import java.util.concurrent.Callable;
-import java.util.function.BiFunction;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -19,6 +19,8 @@
 import io.kroxylicious.proxy.config.ConfigParser;
 import io.kroxylicious.proxy.config.Configuration;
 import io.kroxylicious.proxy.config.PluginFactoryRegistry;
+import io.kroxylicious.proxy.internal.config.Feature;
+import io.kroxylicious.proxy.internal.config.Features;
 
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
@@ -35,13 +37,17 @@ public class Kroxylicious implements Callable<Integer> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger("io.kroxylicious.proxy.StartupShutdownLogger");
     private static final String UNKNOWN = "unknown";
-    private final BiFunction<PluginFactoryRegistry, Configuration, KafkaProxy> proxyBuilder;
+    private final KafkaProxyBuilder proxyBuilder;
+
+    interface KafkaProxyBuilder {
+        KafkaProxy build(PluginFactoryRegistry registry, Configuration config, Features features);
+    }
 
     Kroxylicious() {
         this(KafkaProxy::new);
     }
 
-    Kroxylicious(BiFunction<PluginFactoryRegistry, Configuration, KafkaProxy> proxyBuilder) {
+    Kroxylicious(KafkaProxyBuilder proxyBuilder) {
         this.proxyBuilder = proxyBuilder;
     }
 
@@ -61,8 +67,9 @@ public Integer call() throws Exception {
         try (InputStream stream = Files.newInputStream(configFile.toPath())) {
 
             Configuration config = configParser.parseConfiguration(stream);
-            printBannerAndVersions();
-            try (KafkaProxy kafkaProxy = proxyBuilder.apply(configParser, config)) {
+            Features features = getFeatures();
+            printBannerAndVersions(features);
+            try (KafkaProxy kafkaProxy = proxyBuilder.build(configParser, config, features)) {
                 kafkaProxy.startup();
                 kafkaProxy.block();
             }
@@ -75,12 +82,29 @@ public Integer call() throws Exception {
         return 0;
     }
 
-    private static void printBannerAndVersions() throws Exception {
+    private static boolean isExplicitlyEnabled(Feature feature) {
+        String variableName = "KROXYLICIOUS_UNLOCK_" + feature.name();
+        String enabledString = System.getProperty(variableName, System.getenv(variableName));
+        return Boolean.parseBoolean(enabledString);
+    }
+
+    private static Features getFeatures() {
+        Features.FeaturesBuilder builder = Features.builder();
+        Arrays.stream(Feature.values()).forEach(f -> {
+            if (isExplicitlyEnabled(f)) {
+                builder.enable(f);
+            }
+        });
+        return builder.build();
+    }
+
+    private static void printBannerAndVersions(Features features) throws Exception {
         new BannerLogger().log();
         String[] versions = new VersionProvider().getVersion();
         for (String version : versions) {
             LOGGER.info("{}", version);
         }
+        features.warnings().forEach(LOGGER::warn);
         LOGGER.atInfo()
                 .setMessage("Platform: Java {}({}) running on {} {}/{}")
                 .addArgument(Runtime::version)

kroxylicious-app/src/test/java/io/kroxylicious/app/KroxyliciousIT.java

@@ -6,18 +6,24 @@
 package io.kroxylicious.app;
 
 import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java.util.function.BiFunction;
 
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.junit.jupiter.api.Test;
@@ -26,12 +32,15 @@
 
 import io.kroxylicious.proxy.config.ConfigParser;
 import io.kroxylicious.proxy.config.Configuration;
-import io.kroxylicious.proxy.service.HostPort;
+import io.kroxylicious.proxy.internal.config.Feature;
+import io.kroxylicious.proxy.internal.config.Features;
 import io.kroxylicious.testing.kafka.api.KafkaCluster;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
 import static io.kroxylicious.test.tester.KroxyliciousConfigUtils.proxy;
 import static io.kroxylicious.test.tester.KroxyliciousTesters.kroxyliciousTester;
+import static io.kroxylicious.test.tester.KroxyliciousTesters.newBuilder;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 /**
@@ -45,9 +54,73 @@ class KroxyliciousIT {
     private static final String PLAINTEXT = "Hello, world!";
 
     @Test
-    void shouldProxyWhenRunAsStandaloneProcess(KafkaCluster cluster, Admin admin, @TempDir Path tempDir) throws Exception {
-        var proxyAddress = HostPort.parse("localhost:9192");
+    void shouldFailToStartWithTestConfigurationsByDefault(@TempDir Path tempDir) throws IOException {
+        SubprocessKroxyliciousFactory kroxyliciousFactory = new SubprocessKroxyliciousFactory(tempDir, (features, processBuilder) -> {
+            // no-op so that io is not inherited
+        }, List.of());
+        var tester = kroxyliciousTester(proxy("fake:9092").withDevelopment(Map.of("a", "b")), kroxyliciousFactory);
+        Process lastProcess = kroxyliciousFactory.lastProcess;
+        assertThat(lastProcess).isNotNull();
+        assertThat(lastProcess.onExit()).succeedsWithin(5, TimeUnit.SECONDS);
+        byte[] bytes = lastProcess.getInputStream().readAllBytes();
+        String output = new String(bytes, StandardCharsets.UTF_8);
+        assertThat(output).contains("test-only configuration for proxy present, but loading test-only configuration not enabled");
+        tester.close();
+    }
+
+    @Test
+    void shouldFailToStartWithTestConfigurationAndLoadTestConfigurationExplicitlyDisabled(@TempDir Path tempDir) throws IOException {
+        SubprocessKroxyliciousFactory kroxyliciousFactory = new SubprocessKroxyliciousFactory(tempDir, (features, processBuilder) -> {
+            processBuilder.environment().put(prefixUnlockPropertyName(Feature.TEST_ONLY_CONFIGURATION), "false");
+        }, List.of());
+        var tester = kroxyliciousTester(proxy("fake:9092").withDevelopment(Map.of("a", "b")), kroxyliciousFactory);
+        Process lastProcess = kroxyliciousFactory.lastProcess;
+        assertThat(lastProcess).isNotNull();
+        assertThat(lastProcess.onExit()).succeedsWithin(5, TimeUnit.SECONDS);
+        byte[] bytes = lastProcess.getInputStream().readAllBytes();
+        String output = new String(bytes, StandardCharsets.UTF_8);
+        assertThat(output).contains("test-only configuration for proxy present, but loading test-only configuration not enabled");
+        tester.close();
+    }
+
+    @Test
+    void shouldStartWithTestConfigurationsFeatureEnabledByEnvironmentVariable(KafkaCluster cluster, Admin admin, @TempDir Path tempDir) throws Exception {
+        admin.createTopics(List.of(
+                new NewTopic(TOPIC_1, 1, (short) 1),
+                new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
+
+        try (var tester = newBuilder(proxy(cluster).withDevelopment(Map.of("a", "b")))
+                .setKroxyliciousFactory(new SubprocessKroxyliciousFactory(tempDir))
+                .setFeatures(Features.builder().enable(Feature.TEST_ONLY_CONFIGURATION).build())
+                .createDefaultKroxyliciousTester();
+                var producer = tester.producer(Map.of(
+                        ProducerConfig.CLIENT_ID_CONFIG, "shouldModifyProduceMessage",
+                        ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000));
+                var consumer = tester.consumer()) {
+            assertProxies(producer, consumer);
+        }
+    }
+
+    @Test
+    void shouldStartWithTestConfigurationsFeatureEnabledBySystemProperty(KafkaCluster cluster, Admin admin, @TempDir Path tempDir) throws Exception {
+        admin.createTopics(List.of(
+                new NewTopic(TOPIC_1, 1, (short) 1),
+                new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
 
+        try (var tester = newBuilder(proxy(cluster).withDevelopment(Map.of("a", "b")))
+                .setKroxyliciousFactory(new SubprocessKroxyliciousFactory(tempDir, (features, processBuilder) -> processBuilder.inheritIO(),
+                        List.of("-D" + prefixUnlockPropertyName(Feature.TEST_ONLY_CONFIGURATION) + "=true")))
+                .createDefaultKroxyliciousTester();
+                var producer = tester.producer(Map.of(
+                        ProducerConfig.CLIENT_ID_CONFIG, "shouldModifyProduceMessage",
+                        ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000));
+                var consumer = tester.consumer()) {
+            assertProxies(producer, consumer);
+        }
+    }
+
+    @Test
+    void shouldProxyWhenRunAsStandaloneProcess(KafkaCluster cluster, Admin admin, @TempDir Path tempDir) throws Exception {
         admin.createTopics(List.of(
                 new NewTopic(TOPIC_1, 1, (short) 1),
                 new NewTopic(TOPIC_2, 1, (short) 1))).all().get();
@@ -57,36 +130,70 @@ void shouldProxyWhenRunAsStandaloneProcess(KafkaCluster cluster, Admin admin, @T
                         ProducerConfig.CLIENT_ID_CONFIG, "shouldModifyProduceMessage",
                         ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 3_600_000));
                 var consumer = tester.consumer()) {
-            producer.send(new ProducerRecord<>(TOPIC_1, "my-key", PLAINTEXT)).get();
-            producer.send(new ProducerRecord<>(TOPIC_2, "my-key", PLAINTEXT)).get();
-            producer.flush();
-
-            consumer.subscribe(Set.of(TOPIC_1));
-            ConsumerRecords<String, String> records1 = consumer.poll(Duration.ofSeconds(10));
-            consumer.subscribe(Set.of(TOPIC_2));
-            ConsumerRecords<String, String> records2 = consumer.poll(Duration.ofSeconds(10));
-
-            assertEquals(1, records1.count());
-            assertEquals(PLAINTEXT, records1.iterator().next().value());
-            assertEquals(1, records2.count());
-            assertEquals(PLAINTEXT, records2.iterator().next().value());
+            assertProxies(producer, consumer);
         }
     }
 
-    private record SubprocessKroxyliciousFactory(Path tempDir) implements Function<Configuration, AutoCloseable> {
+    private static void assertProxies(Producer<String, String> producer, Consumer<String, String> consumer)
+            throws InterruptedException, ExecutionException {
+        producer.send(new ProducerRecord<>(KroxyliciousIT.TOPIC_1, "my-key", KroxyliciousIT.PLAINTEXT)).get();
+        producer.send(new ProducerRecord<>(KroxyliciousIT.TOPIC_2, "my-key", KroxyliciousIT.PLAINTEXT)).get();
+        producer.flush();
+
+        consumer.subscribe(Set.of(KroxyliciousIT.TOPIC_1));
+        ConsumerRecords<String, String> records1 = consumer.poll(Duration.ofSeconds(10));
+        consumer.subscribe(Set.of(KroxyliciousIT.TOPIC_2));
+        ConsumerRecords<String, String> records2 = consumer.poll(Duration.ofSeconds(10));
+
+        assertEquals(1, records1.count());
+        assertEquals(KroxyliciousIT.PLAINTEXT, records1.iterator().next().value());
+        assertEquals(1, records2.count());
+        assertEquals(KroxyliciousIT.PLAINTEXT, records2.iterator().next().value());
+    }
+
+    private static String prefixUnlockPropertyName(Feature feature) {
+        return "KROXYLICIOUS_UNLOCK_" + feature.name();
+    }
+
+    private static class SubprocessKroxyliciousFactory implements BiFunction<Configuration, Features, AutoCloseable> {
+
+        private final Path tempDir;
+        private final java.util.function.BiConsumer<Features, ProcessBuilder> processBuilderModifier;
+        private final List<String> jvmArgs;
+        private Process lastProcess;
+
+        SubprocessKroxyliciousFactory(Path tempDir) {
+            this(tempDir, (features, processBuilder) -> {
+                processBuilder.inheritIO();
+                if (features.isEnabled(Feature.TEST_ONLY_CONFIGURATION)) {
+                    processBuilder.environment().put(prefixUnlockPropertyName(Feature.TEST_ONLY_CONFIGURATION), "true");
+                }
+            }, List.of());
+        }
+
+        SubprocessKroxyliciousFactory(Path tempDir, java.util.function.BiConsumer<Features, ProcessBuilder> processBuilderModifier, List<String> jvmArgs) {
+            this.tempDir = tempDir;
+            this.processBuilderModifier = processBuilderModifier;
+            this.jvmArgs = jvmArgs;
+        }
 
         @Override
-        public AutoCloseable apply(Configuration config) {
+        public AutoCloseable apply(Configuration config, Features features) {
             try {
                 Path configPath = tempDir.resolve("config.yaml");
                 Files.writeString(configPath, new ConfigParser().toYaml(config));
                 String java = System.getProperty("java.home") + File.separator + "bin" + File.separator + "java";
                 String classpath = System.getProperty("java.class.path");
-                var processBuilder = new ProcessBuilder(java, "-cp", classpath, "io.kroxylicious.app.Kroxylicious", "-c", configPath.toString()).inheritIO();
-                Process start = processBuilder.start();
+                List<String> command = new ArrayList<>();
+                command.add(java);
+                command.addAll(jvmArgs);
+                command.addAll(List.of("-cp", classpath, "io.kroxylicious.app.Kroxylicious", "-c", configPath.toString()));
+                var processBuilder = new ProcessBuilder(command);
+                processBuilderModifier.accept(features, processBuilder);
+                lastProcess = processBuilder.start();
                 return () -> {
-                    start.destroy();
-                    start.onExit().get(10, TimeUnit.SECONDS);
+                    lastProcess.destroy();
+                    lastProcess.onExit().get(10, TimeUnit.SECONDS);
                 };
             }
             catch (Exception e) {

kroxylicious-app/src/test/java/io/kroxylicious/app/KroxyliciousTest.java

@@ -12,6 +12,7 @@
 import java.io.StringWriter;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -23,6 +24,7 @@
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import io.kroxylicious.proxy.KafkaProxy;
+import io.kroxylicious.proxy.internal.config.Features;
 
 import picocli.CommandLine;
 
@@ -41,10 +43,16 @@ class KroxyliciousTest {
     private CommandLine cmd;
     private StringWriter soutWriter;
     private StringWriter serrWriter;
+    private AtomicReference<Features> features = new AtomicReference<>(null);
 
     @BeforeEach
     public void setup() {
-        Kroxylicious app = new Kroxylicious((ffm, configuration) -> mockProxy);
+        Kroxylicious app = new Kroxylicious((ffm, configuration, features) -> {
+            if (!this.features.compareAndSet(null, features)) {
+                throw new IllegalStateException("env already set");
+            }
+            return mockProxy;
+        });
         soutWriter = new StringWriter();
         serrWriter = new StringWriter();
         cmd = new CommandLine(app);
@@ -87,6 +95,15 @@ void testKroxyliciousStartsAndThenTerminates(@TempDir Path dir) throws Exception
         assertEquals(0, cmd.execute("-c", file.toString()));
     }
 
+    @Test
+    void testDefaultFeatures(@TempDir Path dir) throws Exception {
+        Path file = copyClasspathResourceToTempFileInDir("proxy-config.yaml", dir);
+        when(mockProxy.startup()).thenReturn(mockProxy);
+        doNothing().when(mockProxy).block();
+        assertEquals(0, cmd.execute("-c", file.toString()));
+        assertThat(features).hasValue(Features.defaultFeatures());
+    }
+
     @Test
     void testKroxyliciousExceptionOnBlock(@TempDir Path dir) throws Exception {
         Path file = copyClasspathResourceToTempFileInDir("proxy-config.yaml", dir);

kroxylicious-integration-test-support/src/main/java/io/kroxylicious/test/tester/DefaultKroxyliciousTester.java

@@ -40,6 +40,7 @@
 import io.kroxylicious.proxy.config.ServiceBasedPluginFactoryRegistry;
 import io.kroxylicious.proxy.config.VirtualCluster;
 import io.kroxylicious.proxy.config.tls.Tls;
+import io.kroxylicious.proxy.internal.config.Features;
 import io.kroxylicious.test.client.KafkaClient;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -220,7 +221,7 @@ public <U, V> Consumer<U, V> consumer(String virtualCluster, Serde<U> keySerde,
     public void restartProxy() {
         try {
             proxy.close();
-            proxy = spawnProxy(kroxyliciousConfig);
+            proxy = spawnProxy(kroxyliciousConfig, Features.defaultFeatures());
         }
         catch (Exception e) {
             throw new IllegalStateException(e);
@@ -257,8 +258,8 @@ private static Optional<Exception> closeCloseable(Closeable c) {
         }
     }
 
-    static KafkaProxy spawnProxy(Configuration config) {
-        KafkaProxy kafkaProxy = new KafkaProxy(new ServiceBasedPluginFactoryRegistry(), config);
+    static KafkaProxy spawnProxy(Configuration config, Features features) {
+        KafkaProxy kafkaProxy = new KafkaProxy(new ServiceBasedPluginFactoryRegistry(), config, features);
         try {
             kafkaProxy.startup();
         }