Changes to support Apache Kafka 4.0

* Also upgrades Kroxylicious Junit Ext to 0.11.0

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

Author: k-wall

CHANGELOG.md

@@ -7,6 +7,7 @@ Format `<github issue/pr number>: <short description>`.
 
 ## SNAPSHOT
 
+* [#1855](https://github.com/kroxylicious/kroxylicious/issues/1855) Upgrade to Apache Kafka 4.0
 * [#1928](https://github.com/kroxylicious/kroxylicious/pull/1928) Make Kroxylicious Operator metrics available for collection
 
 ## 0.11.0

kroxylicious-api/pom.xml

@@ -162,6 +162,15 @@
                         <breakBuildBasedOnSemanticVersioningForMajorVersionZero>${ApiCompatability.EnforceForMajorVersionZero}</breakBuildBasedOnSemanticVersioningForMajorVersionZero>
                         <excludes>
                             <exclude>io.kroxylicious.proxy.filter.FilterFactoryContext#eventLoop()</exclude> <!-- https://github.com/kroxylicious/kroxylicious/issues/1380 scheduled removal of deprecated API method -->
+                            <!-- The following filter exclusions relate to RPCs that were removed at Kafka 4.0 -->
+                            <exclude>io.kroxylicious.proxy.filter.ControlledShutdownRequestFilter</exclude>
+                            <exclude>io.kroxylicious.proxy.filter.ControlledShutdownResponseFilter</exclude>
+                            <excludes>io.kroxylicious.proxy.filter.LeaderAndIsrRequestFilter</excludes>
+                            <excludes>io.kroxylicious.proxy.filter.LeaderAndIsrResponseFilter</excludes>
+                            <excludes>io.kroxylicious.proxy.filter.StopReplicaRequestFilter</excludes>
+                            <excludes>io.kroxylicious.proxy.filter.StopReplicaResponseFilter</excludes>
+                            <excludes>io.kroxylicious.proxy.filter.UpdateMetadataRequestFilter</excludes>
+                            <excludes>io.kroxylicious.proxy.filter.UpdateMetadataResponseFilter</excludes>
                         </excludes>
                         <!-- see documentation -->
                     </parameter>

kroxylicious-filter-test-support/src/main/java/io/kroxylicious/test/RequestFactory.java

@@ -15,6 +15,7 @@
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 
 import org.apache.kafka.common.Uuid;
@@ -42,7 +43,6 @@
 import org.apache.kafka.common.message.ReadShareGroupStateRequestData;
 import org.apache.kafka.common.message.ReadShareGroupStateSummaryRequestData;
 import org.apache.kafka.common.message.ShareGroupDescribeRequestData;
-import org.apache.kafka.common.message.UpdateMetadataRequestData;
 import org.apache.kafka.common.message.WriteShareGroupStateRequestData;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.ApiMessage;
@@ -66,6 +66,11 @@ public class RequestFactory {
             ApiKeys.ALTER_REPLICA_LOG_DIRS, ApiKeys.CREATE_PARTITIONS, ApiKeys.ALTER_CLIENT_QUOTAS, ApiKeys.DESCRIBE_USER_SCRAM_CREDENTIALS,
             ApiKeys.ALTER_USER_SCRAM_CREDENTIALS, ApiKeys.DESCRIBE_PRODUCERS, ApiKeys.DESCRIBE_TRANSACTIONS, ApiKeys.DESCRIBE_TOPIC_PARTITIONS);
 
+    /** The following API keys are no longer used by Kafka. They were removed by Kafka 4.0. */
+    @VisibleForTesting
+    protected static final EnumSet<ApiKeys> REMOVED_API_KEYS = EnumSet.of(ApiKeys.CONTROLLED_SHUTDOWN, ApiKeys.LEADER_AND_ISR, ApiKeys.STOP_REPLICA,
+            ApiKeys.UPDATE_METADATA);
+
     @VisibleForTesting
     protected static final Map<ApiKeys, Consumer<ApiMessage>> messagePopulators = new EnumMap<>(ApiKeys.class);
 
@@ -74,7 +79,6 @@ public class RequestFactory {
         messagePopulators.put(ApiKeys.LIST_OFFSETS, RequestFactory::populateListOffsetsRequest);
         messagePopulators.put(ApiKeys.OFFSET_FETCH, RequestFactory::populateOffsetFetchRequest);
         messagePopulators.put(ApiKeys.METADATA, RequestFactory::populateMetadataRequest);
-        messagePopulators.put(ApiKeys.UPDATE_METADATA, RequestFactory::populateUpdateMetadataRequest);
         messagePopulators.put(ApiKeys.LEAVE_GROUP, RequestFactory::populateLeaveGroupRequest);
         messagePopulators.put(ApiKeys.DESCRIBE_GROUPS, RequestFactory::populateDescribeGroupsRequest);
         messagePopulators.put(ApiKeys.CONSUMER_GROUP_DESCRIBE, RequestFactory::populateConsumeGroupDescribeRequest);
@@ -100,7 +104,14 @@ private RequestFactory() {
     }
 
     public static Stream<ApiMessageVersion> apiMessageFor(Function<ApiKeys, Short> versionFunction) {
-        return apiMessageFor(versionFunction, EnumSet.complementOf(RequestFactory.SPECIAL_CASES));
+        return apiMessageFor(versionFunction, supportedApiKeys());
+    }
+
+    @NonNull
+    public static Set<ApiKeys> supportedApiKeys() {
+        var excluded = RequestFactory.SPECIAL_CASES;
+        excluded.addAll(REMOVED_API_KEYS);
+        return EnumSet.complementOf(excluded);
     }
 
     public static Stream<ApiMessageVersion> apiMessageFor(Function<ApiKeys, Short> versionFunction, ApiKeys... apiKeys) {
@@ -115,6 +126,7 @@ public static Stream<ApiMessageVersion> apiMessageFor(Function<ApiKeys, Short> v
     public static Stream<ApiMessageVersion> apiMessageFor(Function<ApiKeys, Short> versionFunction, Set<ApiKeys> apiKeys) {
         return Stream.of(apiKeys)
                 .flatMap(Collection::stream)
+                .filter(Predicate.not(x -> x.messageType.requestSchemas().length == 0 && x.messageType.responseSchemas().length == 0)) // TOOD is there a better way?
                 .map(apiKey -> {
                     final ApiMessage apiMessage = apiMessageForApiKey(apiKey);
                     final Short apiVersion = versionFunction.apply(apiKey);
@@ -176,13 +188,6 @@ private static void populateMetadataRequest(ApiMessage apiMessage) {
         metadataRequestData.setTopics(List.of(t1));
     }
 
-    private static void populateUpdateMetadataRequest(ApiMessage apiMessage) {
-        final UpdateMetadataRequestData updateMetadataRequestData = (UpdateMetadataRequestData) apiMessage;
-        final UpdateMetadataRequestData.UpdateMetadataTopicState t1 = new UpdateMetadataRequestData.UpdateMetadataTopicState();
-        t1.setTopicId(Uuid.randomUuid());
-        updateMetadataRequestData.setTopicStates(List.of(t1));
-    }
-
     private static void populateLeaveGroupRequest(ApiMessage apiMessage) {
         final LeaveGroupRequestData leaveGroupRequestData = (LeaveGroupRequestData) apiMessage;
         final LeaveGroupRequestData.MemberIdentity memberIdentity = new LeaveGroupRequestData.MemberIdentity();

kroxylicious-filter-test-support/src/main/java/io/kroxylicious/test/record/RecordTestUtils.java

@@ -13,7 +13,6 @@
 import java.util.HexFormat;
 import java.util.List;
 
-import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.compress.Compression;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.record.ControlRecordType;
@@ -507,7 +506,7 @@ public static MemoryRecords memoryRecordsWithAllRecordsRemoved(long baseOffset)
             memoryRecordsBuilder.append(DEFAULT_TIMESTAMP, new byte[]{ 1, 2, 3 }, new byte[]{ 1, 2, 3 });
             MemoryRecords records = memoryRecordsBuilder.build();
             ByteBuffer output = ByteBuffer.allocate(1024);
-            records.filterTo(new TopicPartition("any", 1), new MemoryRecords.RecordFilter(DEFAULT_TIMESTAMP, 0L) {
+            MemoryRecords.RecordFilter recordFilter = new MemoryRecords.RecordFilter(DEFAULT_TIMESTAMP, 0L) {
                 @Override
                 protected BatchRetentionResult checkBatchRetention(RecordBatch batch) {
                     return new BatchRetentionResult(BatchRetention.RETAIN_EMPTY, false);
@@ -517,7 +516,8 @@ protected BatchRetentionResult checkBatchRetention(RecordBatch batch) {
                 protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
                     return false;
                 }
-            }, output, 1, BufferSupplier.NO_CACHING);
+            };
+            records.filterTo(recordFilter, output, BufferSupplier.NO_CACHING);
 
             output.flip();
             return MemoryRecords.readableRecords(output);

kroxylicious-filter-test-support/src/test/java/io/kroxylicious/test/RequestFactoryTest.java

@@ -27,7 +27,8 @@ class RequestFactoryTest {
     @Test
     void shouldGenerateRequestForEveryApiKey() {
         // Given
-        final EnumSet<ApiKeys> expectedKeys = EnumSet.complementOf(RequestFactory.SPECIAL_CASES);
+        var expectedKeys = RequestFactory.supportedApiKeys();
+        var absentKeys = EnumSet.complementOf(EnumSet.copyOf(expectedKeys));
 
         // When
         final Stream<RequestFactory.ApiMessageVersion> apiMessages = RequestFactory.apiMessageFor(ApiKeys::latestVersion);
@@ -39,7 +40,7 @@ void shouldGenerateRequestForEveryApiKey() {
                         RequestFactory.ApiMessageVersion::apiMessage));
         assertThat(allMessages).hasSameSizeAs(expectedKeys)
                 .allSatisfy((apiKeys, apiMessage) -> {
-                    assertThat(apiKeys).isNotIn(RequestFactory.SPECIAL_CASES);
+                    assertThat(apiKeys).isNotIn(absentKeys);
                     assertThat(apiMessages).isNotNull();
                 });
     }
@@ -73,4 +74,19 @@ void shouldThrowExceptionIfSpecialCaseRequested(ApiKeys apiKey) {
     static Stream<ApiKeys> specialCases() {
         return RequestFactory.SPECIAL_CASES.stream();
     }
+
+    @ParameterizedTest
+    @MethodSource("removedApiKeys")
+    void shouldThrowExceptionIfRemovedCaseRequested(ApiKeys apiKey) {
+        // Given
+
+        // When
+        assertThatThrownBy(() -> RequestFactory.apiMessageFor(ApiKeys::latestVersion, apiKey)).isInstanceOf(IllegalArgumentException.class);
+
+        // Then
+    }
+
+    static Stream<ApiKeys> removedApiKeys() {
+        return RequestFactory.REMOVED_API_KEYS.stream();
+    }
 }

kroxylicious-filters/kroxylicious-multitenant/src/main/java/io/kroxylicious/proxy/filter/multitenant/MultiTenantFilter.java

@@ -16,6 +16,8 @@
 import org.apache.kafka.common.message.AddPartitionsToTxnRequestData;
 import org.apache.kafka.common.message.AddPartitionsToTxnResponseData;
 import org.apache.kafka.common.message.ApiVersionsResponseData;
+import org.apache.kafka.common.message.ConsumerGroupDescribeRequestData;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
 import org.apache.kafka.common.message.CreateTopicsRequestData;
 import org.apache.kafka.common.message.CreateTopicsResponseData;
 import org.apache.kafka.common.message.DeleteTopicsRequestData;
@@ -60,6 +62,8 @@
 import io.kroxylicious.proxy.filter.AddPartitionsToTxnRequestFilter;
 import io.kroxylicious.proxy.filter.AddPartitionsToTxnResponseFilter;
 import io.kroxylicious.proxy.filter.ApiVersionsResponseFilter;
+import io.kroxylicious.proxy.filter.ConsumerGroupDescribeRequestFilter;
+import io.kroxylicious.proxy.filter.ConsumerGroupDescribeResponseFilter;
 import io.kroxylicious.proxy.filter.CreateTopicsRequestFilter;
 import io.kroxylicious.proxy.filter.CreateTopicsResponseFilter;
 import io.kroxylicious.proxy.filter.DeleteTopicsRequestFilter;
@@ -125,6 +129,7 @@ class MultiTenantFilter
         OffsetDeleteRequestFilter, OffsetDeleteResponseFilter,
         OffsetForLeaderEpochRequestFilter, OffsetForLeaderEpochResponseFilter,
         FindCoordinatorRequestFilter, FindCoordinatorResponseFilter,
+        ConsumerGroupDescribeRequestFilter, ConsumerGroupDescribeResponseFilter,
         ListGroupsResponseFilter,
         JoinGroupRequestFilter,
         SyncGroupRequestFilter,
@@ -324,6 +329,20 @@ public CompletionStage<ResponseFilterResult> onFindCoordinatorResponse(short api
         return context.forwardResponse(header, response);
     }
 
+    @Override
+    public CompletionStage<RequestFilterResult> onConsumerGroupDescribeRequest(short apiVersion, RequestHeaderData header, ConsumerGroupDescribeRequestData request,
+                                                                               FilterContext context) {
+        request.setGroupIds(request.groupIds().stream().map(group -> applyTenantPrefix(context, group)).toList());
+        return context.forwardRequest(header, request);
+    }
+
+    @Override
+    public CompletionStage<ResponseFilterResult> onConsumerGroupDescribeResponse(short apiVersion, ResponseHeaderData header, ConsumerGroupDescribeResponseData response,
+                                                                                 FilterContext context) {
+        response.groups().forEach(group -> removeTenantPrefix(context, group::groupId, group::setGroupId, false));
+        return context.forwardResponse(header, response);
+    }
+
     @Override
     public CompletionStage<ResponseFilterResult> onListGroupsResponse(short apiVersion, ResponseHeaderData header, ListGroupsResponseData response,
                                                                       FilterContext context) {

kroxylicious-filters/kroxylicious-multitenant/src/test/resources/io/kroxylicious/proxy/filter/multitenant/ConsumerGroupDescribe.test.yaml

@@ -0,0 +1,35 @@
+#
+# Copyright Kroxylicious Authors.
+#
+# Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+#
+
+- apiMessageType: CONSUMER_GROUP_DESCRIBE
+  version: 1
+  request:
+    payload:
+      groupIds:
+        - mygroup
+      includeAuthorizedOperations: false
+    diff:
+      - op: replace
+        path: "/groupIds/0"
+        value: tenant1-mygroup
+  response:
+    payload:
+      groups:
+        - groupId: tenant1-mygroup
+          groupState: ""
+          groupEpoch: 0
+          assignmentEpoch: 0
+          assignorName: ""
+          members: []
+          authorizedOperations: 0
+          errorCode: 0
+          errorMessage: ""
+      throttleTimeMs: 0
+    diff:
+      - op: replace
+        path: "/groups/0/groupId"
+        value: mygroup
+  disabled: false

kroxylicious-integration-tests/pom.xml

@@ -196,6 +196,11 @@
             <artifactId>junit-jupiter-params</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.junit-pioneer</groupId>
+            <artifactId>junit-pioneer</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka_2.13</artifactId>

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

@@ -62,7 +62,6 @@
 import io.kroxylicious.testing.kafka.common.BrokerCluster;
 import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
 import io.kroxylicious.testing.kafka.common.SaslMechanism;
-import io.kroxylicious.testing.kafka.common.ZooKeeperCluster;
 import io.kroxylicious.testing.kafka.junit5ext.KafkaClusterExtension;
 
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -638,25 +637,25 @@ void targetClusterDynamicallyAddsBroker(@BrokerCluster KafkaCluster cluster) thr
         }
     }
 
-    // we currently cannot influence the node ids, so we start a 2 node cluster and shutdown node 0
-    // cannot use KRaft as node 0 is a controller
+    // Test extension does not allow us to influence the node ids, so we start a 3 node cluster
+    // and shutdown node 1, leaving us with nodes 0 (controller/broker) and 2 (broker).
     @Test
-    void canConfigurePortIdentifiesNodeWithRange(@ZooKeeperCluster @BrokerCluster(numBrokers = 2) KafkaCluster cluster, Admin admin) throws Exception {
-        cluster.removeBroker(0);
+    void canConfigurePortIdentifiesNodeWithRanges(@BrokerCluster(numBrokers = 3) KafkaCluster cluster, Admin admin) throws Exception {
+        cluster.removeBroker(1);
         await().atMost(Duration.ofSeconds(5)).until(() -> admin.describeCluster().nodes().get(),
-                n -> n.size() == 1 && n.iterator().next().id() == 1);
+                n -> n.stream().map(Node::id).collect(Collectors.toSet()).equals(Set.of(0, 2)));
         var builder = new ConfigurationBuilder()
                 .addToVirtualClusters(KroxyliciousConfigUtils.baseVirtualClusterBuilder(cluster, "demo")
                         .addToGateways(defaultGatewayBuilder()
                                 .withNewPortIdentifiesNode()
                                 .withBootstrapAddress(PROXY_ADDRESS)
-                                .withNodeIdRanges(new NamedRange("myrange", 1, 2))
+                                .withNodeIdRanges(new NamedRange("range1", 0, 0), new NamedRange("range2", 2, 2))
                                 .endPortIdentifiesNode()
                                 .build())
                         .build());
 
         try (var tester = kroxyliciousTester(builder)) {
-            assertThat(cluster.getNumOfBrokers()).isEqualTo(1);
+            assertThat(cluster.getNumOfBrokers()).isEqualTo(2);
             verifyAllBrokersAvailableViaProxy(tester, cluster);
         }
     }

kroxylicious-integration-tests/src/test/java/io/kroxylicious/proxy/filter/oauthbearer/OauthBearerValidationIT.java

@@ -32,6 +32,7 @@
 import org.junit.jupiter.api.condition.EnabledIf;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
+import org.junitpioneer.jupiter.RestoreSystemProperties;
 import org.testcontainers.DockerClientFactory;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.wait.strategy.LogMessageWaitStrategy;
@@ -64,6 +65,7 @@
 @ExtendWith(KafkaClusterExtension.class)
 @ExtendWith(NettyLeakDetectorExtension.class)
 @EnabledIf(value = "isDockerAvailable", disabledReason = "docker unavailable")
+@RestoreSystemProperties
 class OauthBearerValidationIT {
 
     private static final DockerImageName DOCKER_IMAGE_NAME = DockerImageName.parse("ghcr.io/navikt/mock-oauth2-server:2.1.10");
@@ -83,6 +85,7 @@ class OauthBearerValidationIT {
     private static final Predicate<SimpleMetric> DOWNSTREAM_SASL_AUTHENTICATE_PREDICATE = m -> m.name().equals(KROXYLICIOUS_PAYLOAD_SIZE_BYTES_COUNT_METRIC)
             && m.labels().entrySet().containsAll(
                     Map.of("ApiKey", "SASL_AUTHENTICATE", "flowing", "downstream").entrySet());
+    private static final String ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG = "org.apache.kafka.sasl.oauthbearer.allowed.urls";
     @SaslMechanism(value = OAuthBearerLoginModule.OAUTHBEARER_MECHANISM)
     @BrokerConfig(name = "listener.name.external.sasl.oauthbearer.jwks.endpoint.url", value = JWKS_ENDPOINT_URL)
     @BrokerConfig(name = "listener.name.external.sasl.oauthbearer.expected.audience", value = EXPECTED_AUDIENCE)
@@ -92,6 +95,9 @@ class OauthBearerValidationIT {
 
     @BeforeAll
     public static void beforeAll() {
+        // required to permit Kafka to interact with our endpoint.
+        System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, JWKS_ENDPOINT_URL + "," + TOKEN_ENDPOINT_URL);
+
         oauthServer = new OauthServerContainer(OauthBearerValidationIT.DOCKER_IMAGE_NAME);
         oauthServer.setWaitStrategy(new LogMessageWaitStrategy().withRegEx(".*started server on address.*"));
         oauthServer.addFixedExposedPort(OAUTH_SERVER_PORT, OAUTH_SERVER_PORT);
@@ -157,6 +163,8 @@ void authWithBadToken(@TempDir Path tempdir) throws Exception {
         Files.writeString(badTokenFile, BAD_TOKEN);
         var config = getClientConfig(badTokenFile.toUri());
 
+        System.setProperty(ALLOWED_SASL_OAUTHBEARER_URLS_CONFIG, JWKS_ENDPOINT_URL + "," + badTokenFile.toUri());
+
         try (var tester = kroxyliciousTester(getConfiguredProxyBuilder());
                 var admin = tester.admin(config);
                 var ahc = tester.getManagementClient()) {