End to end test verifying a two stage api-versions downgrade actually works for real. (kroxylicious#1662)

Uses Redpanda (22.1.11) as a broker a this version supports ApiVersions version up to 2.  The test uses this broker together with the proxy locked to v3 and a client talking v4.  

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

Author: k-wall

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/ApiVersionsDowngradeIT.java

@@ -23,7 +23,12 @@
 import org.apache.kafka.common.security.plain.PlainLoginModule;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIf;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
+import org.testcontainers.utility.DockerImageName;
 
 import io.github.nettyplus.leakdetector.junit.NettyLeakDetectorExtension;
 import io.netty.buffer.ByteBuf;
@@ -69,6 +74,7 @@ public class ApiVersionsDowngradeIT {
     public static final short API_VERSIONS_ID = ApiKeys.API_VERSIONS.id;
     public static final String SASL_USER = "alice";
     public static final String SASL_PASSWORD = "foo";
+    private static final DockerImageName OLD_REDPANDA_USING_API_VER0_2 = DockerImageName.parse("docker.redpanda.com/redpandadata/redpanda:v22.1.11");
 
     @Test
     void clientAheadOfProxy() {
@@ -136,6 +142,51 @@ private void doProxyRestrictedToOlderApiVersion(KafkaCluster cluster, Map<String
         }
     }
 
+    /**
+     * In this test, we verify the assumption that the Kafka Client is capable
+     * of negotiating down api-versions version twice.  This test uses an
+     * old version of Redpanda (that supports max v2) and the Proxy (restricted
+     * to v3).  The Kafka Client is using v4.
+     * <br>
+     * Client will first make a v4 request.  The proxy's response will cause the client
+     * to try again at v3.  The broker will then cause the client to try a third time at v2.
+     * This request will satisfy all parties and the connection establishment will continue.
+     * All this occurs on a single connection.
+     */
+    @Test
+    @EnabledIf(value = "isDockerAvailable", disabledReason = "docker unavailable")
+    void clientAheadOfProxyWhichIsAheadOfBroker() {
+
+        try (var redpanda = createRedpanda(OLD_REDPANDA_USING_API_VER0_2)) {
+            redpanda.start();
+            var redpandaApiVersion = (short) 2;
+            var proxyApiVersion = (short) (ApiKeys.API_VERSIONS.latestVersion() - 1);
+            assertThat(redpandaApiVersion).isLessThan(proxyApiVersion);
+
+            var testConfigEnabled = Features.builder().enable(Feature.TEST_ONLY_CONFIGURATION).build();
+            var proxy = proxy("localhost:9092")
+                    .withDevelopment(Map.of("apiKeyIdMaxVersionOverride", Map.of(ApiKeys.API_VERSIONS.name(), proxyApiVersion)));
+            try (var tester = newBuilder(proxy).setFeatures(testConfigEnabled).createDefaultKroxyliciousTester();
+                    var admin = tester.admin()) {
+                // We've got no way to observe the actual version of the API versions request that is used during _negotiation_
+                // so we make do with asserting the connection is usable.
+                final var result = admin.describeCluster().clusterId();
+                assertThat(result).as("Unable to get the clusterId from the Kafka cluster").succeedsWithin(Duration.ofSeconds(10));
+                // check that the client is actually using the correct version.
+                assertThat(admin)
+                        .extracting("instance")
+                        .extracting("client")
+                        .extracting("apiVersions")
+                        .extracting("nodeApiVersions", InstanceOfAssertFactories.map(String.class, NodeApiVersions.class))
+                        .hasEntrySatisfying("-1", nav -> {
+                            assertThat(nav.apiVersion(ApiKeys.API_VERSIONS).maxVersion())
+                                    .isEqualTo(redpandaApiVersion);
+                        });
+            }
+
+        }
+    }
+
     private static @NonNull OpaqueRequestFrame createHypotheticalFutureRequest() {
         short unsupportedVersion = (short) (ApiKeys.API_VERSIONS.latestVersion(true) + 1);
         RequestHeaderData requestHeaderData = getRequestHeaderData(API_VERSIONS_ID, unsupportedVersion, CORRELATION_ID);
@@ -164,4 +215,27 @@ private void doProxyRestrictedToOlderApiVersion(KafkaCluster cluster, Map<String
         return requestHeaderData;
     }
 
+    @NonNull
+    private RedpandaContainer createRedpanda(DockerImageName image) {
+        var redpanda = new RedpandaContainer(image);
+        redpanda.setWaitStrategy(new LogMessageWaitStrategy().withRegEx(".*Bootstrap complete.*"));
+        redpanda.addFixedExposedPort(9092, 9092);
+        return redpanda;
+    }
+
+    private static class RedpandaContainer extends GenericContainer<RedpandaContainer> {
+
+        private RedpandaContainer(DockerImageName dockerImageName) {
+            super(dockerImageName);
+        }
+
+        @Override
+        protected void addFixedExposedPort(int hostPort, int containerPort) {
+            super.addFixedExposedPort(hostPort, containerPort);
+        }
+    }
+
+    static boolean isDockerAvailable() {
+        return DockerClientFactory.instance().isDockerAvailable();
+    }
 }