Merge pull request kroxylicious#1648 from robobario/development-mode

Add opt-in test-only configuration to Proxy

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();
         }