Merge branch 'trunk' into KAFKA-20167

Author: m1a2st

.github/workflows/docker_scan.yml

@@ -26,7 +26,7 @@ jobs:
     strategy:
       matrix:
         # This is an array of supported tags. Make sure this array only contains the supported tags
-        supported_image_tag: ['latest', '3.9.1', '4.0.1', '4.1.1', '4.2.0']
+        supported_image_tag: ['latest', '3.9.2', '4.0.1', '4.1.1', '4.2.0']
     steps:
       - name: Run CVE scan
         uses: aquasecurity/trivy-action@b6643a29fecd7f34b3597bc6acb0a98b03d33ff8 # v0.33.1

LICENSE-binary

@@ -225,16 +225,16 @@ License Version 2.0:
 - jakarta.inject-api-2.0.1
 - jakarta.validation-api-3.0.2
 - javassist-3.30.2-GA
-- jetty-alpn-client-12.0.32
-- jetty-client-12.0.32
-- jetty-ee10-servlet-12.0.32
-- jetty-ee10-servlets-12.0.32
-- jetty-http-12.0.32
-- jetty-io-12.0.32
-- jetty-security-12.0.32
-- jetty-server-12.0.32
-- jetty-session-12.0.32
-- jetty-util-12.0.32
+- jetty-alpn-client-12.0.25
+- jetty-client-12.0.25
+- jetty-ee10-servlet-12.0.25
+- jetty-ee10-servlets-12.0.25
+- jetty-http-12.0.25
+- jetty-io-12.0.25
+- jetty-security-12.0.25
+- jetty-server-12.0.25
+- jetty-session-12.0.25
+- jetty-util-12.0.25
 - jose4j-0.9.6
 - jspecify-1.0.0
 - log4j-api-2.25.3

build.gradle

@@ -39,7 +39,7 @@ plugins {
 
   id "com.github.spotbugs" version '6.4.4' apply false
   id 'org.scoverage' version '8.1' apply false
-  id 'com.gradleup.shadow' version '8.3.9' apply false
+  id 'com.gradleup.shadow' version '9.3.1' apply false
   id 'com.diffplug.spotless' version "8.0.0"
 }
 

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

@@ -39,6 +39,10 @@ public LogDirDescription(ApiException error, Map<TopicPartition, ReplicaInfo> re
         this(error, replicaInfos, UNKNOWN_VOLUME_BYTES, UNKNOWN_VOLUME_BYTES, false);
     }
 
+    public LogDirDescription(ApiException error, Map<TopicPartition, ReplicaInfo> replicaInfos, long totalBytes, long usableBytes) {
+        this(error, replicaInfos, totalBytes, usableBytes, false);
+    }
+
     public LogDirDescription(ApiException error, Map<TopicPartition, ReplicaInfo> replicaInfos, long totalBytes, long usableBytes, boolean isCordoned) {
         this.error = error;
         this.replicaInfos = replicaInfos;

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -3172,6 +3172,12 @@ class KafkaApis(val requestChannel: RequestChannel,
 
     val groupId = shareFetchRequest.data.groupId
 
+    if (groupId == null) {
+      requestHelper.sendMaybeThrottle(request,
+        shareFetchRequest.getErrorResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, Errors.INVALID_REQUEST.exception("Invalid group id in the request.")))
+      return CompletableFuture.completedFuture[Unit](())
+    }
+
     // Share Fetch needs permission to perform the READ action on the named group resource (groupId)
     if (!authHelper.authorize(request.context, READ, GROUP, groupId)) {
       requestHelper.sendMaybeThrottle(request, shareFetchRequest.getErrorResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, Errors.GROUP_AUTHORIZATION_FAILED.exception))
@@ -3539,6 +3545,12 @@ class KafkaApis(val requestChannel: RequestChannel,
 
     val groupId = shareAcknowledgeRequest.data.groupId
 
+    if (groupId == null) {
+      requestHelper.sendMaybeThrottle(request,
+        shareAcknowledgeRequest.getErrorResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, Errors.INVALID_REQUEST.exception("Invalid group id in the request.")))
+      return CompletableFuture.completedFuture[Unit](())
+    }
+
     // Share Acknowledge needs permission to perform READ action on the named group resource (groupId)
     if (!authHelper.authorize(request.context, READ, GROUP, groupId)) {
       requestHelper.sendMaybeThrottle(request,
@@ -4222,7 +4234,7 @@ class KafkaApis(val requestChannel: RequestChannel,
    * @return boolean if the member id in the RPC is valid or not.
    */
   def isMemberIdValid(memberId: String): Boolean = {
-    memberId.nonEmpty && memberId.length <= 36
+    memberId != null && memberId.nonEmpty && memberId.length <= 36
   }
 
   private def updateRecordConversionStats(request: RequestChannel.Request,

core/src/test/scala/unit/kafka/log/LogCleanerLagIntegrationTest.scala

@@ -1070,42 +1070,42 @@ class DynamicBrokerConfigTest {
     // Cordoning 1 new log dir, so 1 new handleCordoned invocation
     val props = new Properties()
     props.put(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, logDirs.get(0))
-    ctx.config.dynamicConfig.updateDefaultConfig(props)
+    ctx.config.dynamicConfig.updateBrokerConfig(0, props)
     assertEquals(util.List.of(logDirs.get(0)), ctx.config.cordonedLogDirs)
     verify(ctx.directoryEventHandler, times(1)).handleCordoned(anySet)
     verify(ctx.directoryEventHandler, never()).handleUncordoned(anySet)
 
     // When using *, no other entries must be specified, so no new invocations
     props.put(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, "*,/invalid/log/dir")
-    ctx.config.dynamicConfig.updateDefaultConfig(props)
+    ctx.config.dynamicConfig.updateBrokerConfig(0, props)
     assertEquals(util.List.of(logDirs.get(0)), ctx.config.cordonedLogDirs)
     verify(ctx.directoryEventHandler, times(1)).handleCordoned(anySet)
     verify(ctx.directoryEventHandler, never()).handleUncordoned(anySet)
 
     // Invalid log dir, so no new invocations
     props.put(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, "/invalid/log/dir")
-    ctx.config.dynamicConfig.updateDefaultConfig(props)
+    ctx.config.dynamicConfig.updateBrokerConfig(0, props)
     assertEquals(util.List.of(logDirs.get(0)), ctx.config.cordonedLogDirs)
     verify(ctx.directoryEventHandler, times(1)).handleCordoned(anySet)
     verify(ctx.directoryEventHandler, times(0)).handleUncordoned(anySet)
 
     // * cordons the 2nd log dir, so 1 new handleCordoned invocation
     props.put(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, "*")
-    ctx.config.dynamicConfig.updateDefaultConfig(props)
+    ctx.config.dynamicConfig.updateBrokerConfig(0, props)
     assertEquals(logDirs, ctx.config.cordonedLogDirs)
     verify(ctx.directoryEventHandler, times(2)).handleCordoned(anySet)
     verify(ctx.directoryEventHandler, never()).handleUncordoned(anySet)
 
     // clearing all cordoned log dirs, so 1 new handleUncordoned invocation
     props.put(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, "")
-    ctx.config.dynamicConfig.updateDefaultConfig(props)
+    ctx.config.dynamicConfig.updateBrokerConfig(0, props)
     assertTrue(ctx.config.cordonedLogDirs.isEmpty)
     verify(ctx.directoryEventHandler, times(2)).handleCordoned(anySet)
     verify(ctx.directoryEventHandler, times(1)).handleUncordoned(anySet)
 
     // * cordons all log dirs, so 1 new handleCordoned invocation
     props.put(ServerLogConfigs.CORDONED_LOG_DIRS_CONFIG, String.join(",", logDirs))
-    ctx.config.dynamicConfig.updateDefaultConfig(props)
+    ctx.config.dynamicConfig.updateBrokerConfig(0, props)
     assertEquals(logDirs, ctx.config.cordonedLogDirs)
     verify(ctx.directoryEventHandler, times(3)).handleCordoned(anySet)
     verify(ctx.directoryEventHandler, times(1)).handleUncordoned(anySet)

core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala

@@ -329,6 +329,7 @@ class RequestQuotaTest extends BaseRequestTest {
                 )
               )
           )
+
         case ApiKeys.OFFSET_FETCH =>
           OffsetFetchRequest.Builder.forTopicNames(
             new OffsetFetchRequestData()
@@ -493,6 +494,7 @@ class RequestQuotaTest extends BaseRequestTest {
               .setHost("*")
               .setOperation(AclOperation.WRITE.code)
               .setPermissionType(AclPermissionType.DENY.code))))
+
         case ApiKeys.DELETE_ACLS =>
           new DeleteAclsRequest.Builder(new DeleteAclsRequestData().setFilters(util.List.of(
             new DeleteAclsRequestData.DeleteAclsFilter()
@@ -503,6 +505,7 @@ class RequestQuotaTest extends BaseRequestTest {
               .setHostFilter("*")
               .setOperation(AclOperation.ANY.code)
               .setPermissionType(AclPermissionType.DENY.code))))
+
         case ApiKeys.DESCRIBE_CONFIGS =>
           new DescribeConfigsRequest.Builder(new DescribeConfigsRequestData()
             .setResources(util.List.of(new DescribeConfigsRequestData.DescribeConfigsResource()
@@ -719,7 +722,10 @@ class RequestQuotaTest extends BaseRequestTest {
                   ).iterator)))
 
         case ApiKeys.SHARE_ACKNOWLEDGE =>
-          new ShareAcknowledgeRequest.Builder(new ShareAcknowledgeRequestData())
+          new ShareAcknowledgeRequest.Builder(
+            new ShareAcknowledgeRequestData()
+              .setGroupId("test-share-group")
+              .setMemberId(Uuid.randomUuid().toString))
 
         case ApiKeys.ADD_RAFT_VOTER =>
           new AddRaftVoterRequest.Builder(new AddRaftVoterRequestData())

core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala

@@ -52,7 +52,7 @@ Kafka is a partitioned system so not all servers have the complete data set. Ins
 
 All systems of this nature have the question of how a particular piece of data is assigned to a particular partition. Kafka clients directly control this assignment, the brokers themselves enforce no particular semantics of which messages should be published to a particular partition. Rather, to publish messages the client directly addresses messages to a particular partition, and when fetching messages, fetches from a particular partition. If two clients want to use the same partitioning scheme they must use the same method to compute the mapping of key to partition.
 
-These requests to publish or fetch data must be sent to the broker that is currently acting as the leader for a given partition. This condition is enforced by the broker, so a request for a particular partition to the wrong broker will result in an the NotLeaderForPartition error code (described below).
+These requests to publish or fetch data must be sent to the broker that is currently acting as the leader for a given partition. This condition is enforced by the broker, so a request for a particular partition to the wrong broker will result in the NotLeaderForPartition error code (described below).
 
 How can the client find out which topics exist, what partitions they have, and which brokers currently host those partitions so that it can direct its requests to the right hosts? This information is dynamic, so you can't just configure each client with some static mapping file. Instead all Kafka brokers can answer a metadata request that describes the current state of the cluster: what topics there are, which partitions those topics have, which broker is the leader for those partitions, and the host and port information for these brokers.
 
@@ -79,7 +79,7 @@ Partitioning really serves two purposes in Kafka:
 
 For a given use case you may care about only one of these or both.
 
-To accomplish simple load balancing a simple approach would be for the client to just round robin requests over all brokers. Another alternative, in an environment where there are many more producers than brokers, would be to have each client chose a single partition at random and publish to that. This later strategy will result in far fewer TCP connections.
+To accomplish simple load balancing a simple approach would be for the client to just round robin requests over all brokers. Another alternative, in an environment where there are many more producers than brokers, would be to have each client choose a single partition at random and publish to that. This latter strategy will result in far fewer TCP connections.
 
 Semantic partitioning means using some key in the message to assign messages to partitions. For example if you were processing a click message stream you might want to partition the stream by the user id so that all data for a particular user would go to a single consumer. To accomplish this the client can take a key associated with the message and use some hash of this key to choose the partition to which to deliver the message.
 
@@ -113,7 +113,7 @@ The following sequence may be used by a client to obtain supported API versions
   2. On receiving `ApiVersionsRequest`, a broker returns its full list of supported ApiKeys and versions regardless of current authentication state (e.g., before SASL authentication on an SASL listener, do note that no Kafka protocol requests may take place on an SSL listener before the SSL handshake is finished). If this is considered to leak information about the broker version a workaround is to use SSL with client authentication which is performed at an earlier stage of the connection where the `ApiVersionRequest` is not available. Also, note that broker versions older than 0.10.0.0 do not support this API and will either ignore the request or close connection in response to the request. Also note that if the client `ApiVersionsRequest` version is unsupported by the broker (client is ahead), and the broker version is 2.4.0 or greater, then the broker will respond with a version 0 ApiVersionsResponse with the error code set to `UNSUPPORTED_VERSION` and the `api_versions` field populated with the supported version of the `ApiVersionsRequest`. It is then up to the client to retry, making another `ApiVersionsRequest` using the highest version supported by the client and broker. See [KIP-511: Collect and Expose Client's Name and Version in the Brokers](https://cwiki.apache.org/confluence/x/qRJ4Bw)
   3. If multiple versions of an API are supported by broker and client, clients are recommended to use the latest version supported by the broker and itself.
   4. Deprecation of a protocol version is done by marking an API version as deprecated in the protocol documentation.
-  5. Supported API versions obtained from a broker are only valid for the connection on which that information is obtained. In the event of disconnection, the client should obtain the information from the broker again, as the broker might have been upgraded/downgraded in the mean time.
+  5. Supported API versions obtained from a broker are only valid for the connection on which that information is obtained. In the event of disconnection, the client should obtain the information from the broker again, as the broker might have been upgraded/downgraded in the meantime.