KAFKA-18601: Assume a baseline of 3.3 for server protocol versions

clients/src/main/resources/common/message/AlterPartitionRequest.json

@@ -18,12 +18,13 @@
   "type": "request",
   "listeners": ["controller"],
   "name": "AlterPartitionRequest",
+  // Versions 0-1 were removed in Apache Kafka 4.0, version 2 is the new baseline.
+
   // Version 1 adds LeaderRecoveryState field (KIP-704).
-  //
   // Version 2 adds TopicId field to replace TopicName field (KIP-841).
   //
   // Version 3 adds the NewIsrEpochs field and deprecates the NewIsr field (KIP-903).
-  "validVersions": "0-3",
+  "validVersions": "2-3",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
@@ -32,8 +33,6 @@
       "about": "The epoch of the requesting broker." },
     { "name": "Topics", "type": "[]TopicData", "versions": "0+",
       "about": "The topics to alter ISRs for.", "fields": [
-      { "name":  "TopicName", "type": "string", "versions": "0-1", "ignorable": true, "entityType": "topicName",
-        "about": "The name of the topic to alter ISRs for." },
       { "name":  "TopicId", "type": "uuid", "versions": "2+", "ignorable": true,
         "about": "The ID of the topic to alter ISRs for." },
       { "name": "Partitions", "type": "[]PartitionData", "versions": "0+",

clients/src/main/resources/common/message/AlterPartitionResponse.json

@@ -17,13 +17,14 @@
   "apiKey": 56,
   "type": "response",
   "name": "AlterPartitionResponse",
+  // Versions 0-1 were removed in Apache Kafka 4.0, version 2 is the new baseline.
+
   // Version 1 adds LeaderRecoveryState field (KIP-704).
-  //
   // Version 2 adds TopicId field to replace TopicName field, can return the following new errors:
   // INELIGIBLE_REPLICA, NEW_LEADER_ELECTED and UNKNOWN_TOPIC_ID (KIP-841).
   //
   // Version 3 is the same as version 2 (KIP-903).
-  "validVersions": "0-3",
+  "validVersions": "2-3",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
@@ -32,8 +33,6 @@
       "about": "The top level response error code." },
     { "name": "Topics", "type": "[]TopicData", "versions": "0+",
       "about": "The responses for each topic.", "fields": [
-      { "name": "TopicName", "type": "string", "versions": "0-1", "ignorable": true, "entityType": "topicName",
-        "about": "The name of the topic." },
       { "name":  "TopicId", "type": "uuid", "versions": "2+", "ignorable": true,
         "about": "The ID of the topic." },
       { "name": "Partitions", "type": "[]PartitionData", "versions": "0+",

clients/src/test/java/org/apache/kafka/common/requests/AlterPartitionRequestTest.java

@@ -34,7 +34,6 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class AlterPartitionRequestTest {
-    String topic = "test-topic";
     Uuid topicId = Uuid.randomUuid();
 
     @ParameterizedTest
@@ -44,9 +43,7 @@ public void testBuildAlterPartitionRequest(short version) {
             .setBrokerId(1)
             .setBrokerEpoch(1);
 
-        TopicData topicData = new TopicData()
-            .setTopicId(topicId)
-            .setTopicName(topic);
+        TopicData topicData = new TopicData().setTopicId(topicId);
 
         List<BrokerState> newIsrWithBrokerEpoch = new LinkedList<>();
         newIsrWithBrokerEpoch.add(new BrokerState().setBrokerId(1).setBrokerEpoch(1001));

clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java

@@ -1097,7 +1097,7 @@ private AbstractResponse getResponse(ApiKeys apikey, short version) {
             case BEGIN_QUORUM_EPOCH: return createBeginQuorumEpochResponse();
             case END_QUORUM_EPOCH: return createEndQuorumEpochResponse();
             case DESCRIBE_QUORUM: return createDescribeQuorumResponse();
-            case ALTER_PARTITION: return createAlterPartitionResponse(version);
+            case ALTER_PARTITION: return createAlterPartitionResponse();
             case UPDATE_FEATURES: return createUpdateFeaturesResponse();
             case ENVELOPE: return createEnvelopeResponse();
             case FETCH_SNAPSHOT: return createFetchSnapshotResponse();
@@ -1695,41 +1695,31 @@ private AlterPartitionRequest createAlterPartitionRequest(short version) {
             .setPartitionIndex(1)
             .setPartitionEpoch(2)
             .setLeaderEpoch(3)
-            .setNewIsrWithEpochs(AlterPartitionRequest.newIsrToSimpleNewIsrWithBrokerEpochs(asList(1, 2)));
-
-        if (version >= 1) {
-            // Use the none default value; 1 - RECOVERING
-            partitionData.setLeaderRecoveryState((byte) 1);
-        }
+            .setNewIsrWithEpochs(AlterPartitionRequest.newIsrToSimpleNewIsrWithBrokerEpochs(asList(1, 2)))
+            .setLeaderRecoveryState((byte) 1); // non-default value
 
         AlterPartitionRequestData data = new AlterPartitionRequestData()
             .setBrokerEpoch(123L)
             .setBrokerId(1)
             .setTopics(singletonList(new AlterPartitionRequestData.TopicData()
-                .setTopicName("topic1")
                 .setTopicId(Uuid.randomUuid())
                 .setPartitions(singletonList(partitionData))));
         return new AlterPartitionRequest.Builder(data).build(version);
     }
 
-    private AlterPartitionResponse createAlterPartitionResponse(int version) {
+    private AlterPartitionResponse createAlterPartitionResponse() {
         AlterPartitionResponseData.PartitionData partitionData = new AlterPartitionResponseData.PartitionData()
             .setPartitionEpoch(1)
             .setIsr(asList(0, 1, 2))
             .setErrorCode(Errors.NONE.code())
             .setLeaderEpoch(2)
-            .setLeaderId(3);
-
-        if (version >= 1) {
-            // Use the none default value; 1 - RECOVERING
-            partitionData.setLeaderRecoveryState((byte) 1);
-        }
+            .setLeaderId(3)
+            .setLeaderRecoveryState((byte) 1); // non-default value
 
         AlterPartitionResponseData data = new AlterPartitionResponseData()
                 .setErrorCode(Errors.NONE.code())
                 .setThrottleTimeMs(123)
                 .setTopics(singletonList(new AlterPartitionResponseData.TopicData()
-                    .setTopicName("topic1")
                     .setTopicId(Uuid.randomUuid())
                     .setPartitions(singletonList(partitionData))));
         return new AlterPartitionResponse(data);

core/src/main/scala/kafka/cluster/Partition.scala

@@ -118,8 +118,8 @@ object Partition {
             time: Time,
             replicaManager: ReplicaManager): Partition = {
     Partition(
-      topicPartition = topicIdPartition.topicPartition(),
-      topicId = Option(topicIdPartition.topicId()),
+      topicPartition = topicIdPartition.topicPartition,
+      topicId = Some(topicIdPartition.topicId),
       time = time,
       replicaManager = replicaManager)
   }
@@ -1813,7 +1813,7 @@ class Partition(val topicPartition: TopicPartition,
   private def submitAlterPartition(proposedIsrState: PendingPartitionChange): CompletableFuture[LeaderAndIsr] = {
     debug(s"Submitting ISR state change $proposedIsrState")
     val future = alterIsrManager.submit(
-      new TopicIdPartition(topicId.getOrElse(Uuid.ZERO_UUID), topicPartition),
+      new org.apache.kafka.server.common.TopicIdPartition(topicId.getOrElse(throw new IllegalStateException("Topic id not set for " + topicPartition)), topicPartition.partition),
       proposedIsrState.sentLeaderAndIsr
     )
     future.whenComplete { (leaderAndIsr, e) =>

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

@@ -16,23 +16,18 @@
  */
 package kafka.server
 
-import java.util
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.{CompletableFuture, ConcurrentHashMap}
 import kafka.utils.Logging
 import org.apache.kafka.clients.ClientResponse
-import org.apache.kafka.common.TopicIdPartition
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.Uuid
 import org.apache.kafka.common.errors.OperationNotAttemptedException
 import org.apache.kafka.common.message.AlterPartitionRequestData
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.RequestHeader
 import org.apache.kafka.common.requests.{AlterPartitionRequest, AlterPartitionResponse}
 import org.apache.kafka.common.utils.Time
 import org.apache.kafka.metadata.{LeaderAndIsr, LeaderRecoveryState}
-import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, MetadataVersion, NodeToControllerChannelManager}
+import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, NodeToControllerChannelManager, TopicIdPartition}
 import org.apache.kafka.server.util.Scheduler
 
 import scala.collection.mutable
@@ -92,8 +87,7 @@ object AlterPartitionManager {
       scheduler = scheduler,
       time = time,
       brokerId = config.brokerId,
-      brokerEpochSupplier = brokerEpochSupplier,
-      metadataVersionSupplier = () => metadataCache.metadataVersion()
+      brokerEpochSupplier = brokerEpochSupplier
     )
   }
 }
@@ -104,7 +98,6 @@ class DefaultAlterPartitionManager(
   val time: Time,
   val brokerId: Int,
   val brokerEpochSupplier: () => Long,
-  val metadataVersionSupplier: () => MetadataVersion
 ) extends AlterPartitionManager with Logging {
 
   // Used to allow only one pending ISR update per partition (visible for testing).
@@ -114,7 +107,7 @@ class DefaultAlterPartitionManager(
   // and re-created, we cannot have two entries in this Map especially if we cannot 
   // use an AlterPartition request version which supports topic ids in the end because 
   // the two updates with the same name would be merged together.
-  private[server] val unsentIsrUpdates: util.Map[TopicPartition, AlterPartitionItem] = new ConcurrentHashMap[TopicPartition, AlterPartitionItem]()
+  private[server] val unsentIsrUpdates = new ConcurrentHashMap[TopicIdPartition, AlterPartitionItem]()
 
   // Used to allow only one in-flight request at a time
   private val inflightRequest: AtomicBoolean = new AtomicBoolean(false)
@@ -133,7 +126,7 @@ class DefaultAlterPartitionManager(
   ): CompletableFuture[LeaderAndIsr] = {
     val future = new CompletableFuture[LeaderAndIsr]()
     val alterPartitionItem = AlterPartitionItem(topicIdPartition, leaderAndIsr, future)
-    val enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition.topicPartition, alterPartitionItem) == null
+    val enqueued = unsentIsrUpdates.putIfAbsent(alterPartitionItem.topicIdPartition, alterPartitionItem) == null
     if (enqueued) {
       maybePropagateIsrChanges()
     } else {
@@ -161,7 +154,7 @@ class DefaultAlterPartitionManager(
 
   private def sendRequest(inflightAlterPartitionItems: Seq[AlterPartitionItem]): Unit = {
     val brokerEpoch = brokerEpochSupplier()
-    val (request, topicNamesByIds) = buildRequest(inflightAlterPartitionItems, brokerEpoch)
+    val request = buildRequest(inflightAlterPartitionItems, brokerEpoch)
     debug(s"Sending AlterPartition to controller $request")
 
     // We will not timeout AlterPartition request, instead letting it retry indefinitely
@@ -182,11 +175,9 @@ class DefaultAlterPartitionManager(
               Errors.UNSUPPORTED_VERSION
             } else {
               handleAlterPartitionResponse(
-                response.requestHeader,
                 response.responseBody.asInstanceOf[AlterPartitionResponse],
                 brokerEpoch,
-                inflightAlterPartitionItems,
-                topicNamesByIds
+                inflightAlterPartitionItems
               )
             }
           } finally {
@@ -218,59 +209,40 @@ class DefaultAlterPartitionManager(
    * supported by the controller. The final decision is taken when the AlterPartitionRequest
    * is built in the network client based on the advertised api versions of the controller.
    *
-   * @return A tuple containing the AlterPartitionRequest.Builder and a mapping from
-   *         topic id to topic name. This mapping is used in the response handling.
+   * @return an AlterPartitionRequest.Builder with the provided parameters.
    */
   private def buildRequest(
     inflightAlterPartitionItems: Seq[AlterPartitionItem],
     brokerEpoch: Long
-  ): (AlterPartitionRequest.Builder, mutable.Map[Uuid, String]) = {
-    val metadataVersion = metadataVersionSupplier()
-    // We build this mapping in order to map topic id back to their name when we
-    // receive the response. We cannot rely on the metadata cache for this because
-    // the metadata cache is updated after the partition state so it might not know
-    // yet about a topic id already used here.
-    val topicNamesByIds = mutable.HashMap[Uuid, String]()
-
+  ): AlterPartitionRequest.Builder = {
     val message = new AlterPartitionRequestData()
       .setBrokerId(brokerId)
       .setBrokerEpoch(brokerEpoch)
 
-    inflightAlterPartitionItems.groupBy(_.topicIdPartition.topic).foreach { case (topicName, items) =>
-      val topicId = items.head.topicIdPartition.topicId
-      topicNamesByIds(topicId) = topicName
-
-      // Both the topic name and the topic id are set here because at this stage
-      // we don't know which version of the request will be used.
-      val topicData = new AlterPartitionRequestData.TopicData()
-        .setTopicName(topicName)
-        .setTopicId(topicId)
+    inflightAlterPartitionItems.groupBy(_.topicIdPartition.topicId).foreach { case (topicId, items) =>
+      val topicData = new AlterPartitionRequestData.TopicData().setTopicId(topicId)
       message.topics.add(topicData)
 
       items.foreach { item =>
         val partitionData = new AlterPartitionRequestData.PartitionData()
-          .setPartitionIndex(item.topicIdPartition.partition)
+          .setPartitionIndex(item.topicIdPartition.partitionId)
           .setLeaderEpoch(item.leaderAndIsr.leaderEpoch)
           .setNewIsrWithEpochs(item.leaderAndIsr.isrWithBrokerEpoch)
           .setPartitionEpoch(item.leaderAndIsr.partitionEpoch)
 
-        if (metadataVersion.isLeaderRecoverySupported) {
-          partitionData.setLeaderRecoveryState(item.leaderAndIsr.leaderRecoveryState.value)
-        }
+        partitionData.setLeaderRecoveryState(item.leaderAndIsr.leaderRecoveryState.value)
 
         topicData.partitions.add(partitionData)
       }
     }
 
-    (new AlterPartitionRequest.Builder(message), topicNamesByIds)
+    new AlterPartitionRequest.Builder(message)
   }
 
   private def handleAlterPartitionResponse(
-    requestHeader: RequestHeader,
     alterPartitionResp: AlterPartitionResponse,
     sentBrokerEpoch: Long,
     inflightAlterPartitionItems: Seq[AlterPartitionItem],
-    topicNamesByIds: mutable.Map[Uuid, String]
   ): Errors = {
     val data = alterPartitionResp.data
 
@@ -284,37 +256,31 @@ class DefaultAlterPartitionManager(
 
       case Errors.NONE =>
         // Collect partition-level responses to pass to the callbacks
-        val partitionResponses = new mutable.HashMap[TopicPartition, Either[Errors, LeaderAndIsr]]()
+        val partitionResponses = new mutable.HashMap[TopicIdPartition, Either[Errors, LeaderAndIsr]]()
         data.topics.forEach { topic =>
-          // Topic IDs are used since version 2 of the AlterPartition API.
-          val topicName = if (requestHeader.apiVersion > 1) topicNamesByIds.get(topic.topicId).orNull else topic.topicName
-          if (topicName == null || topicName.isEmpty) {
-            error(s"Received an unexpected topic $topic in the alter partition response, ignoring it.")
-          } else {
-            topic.partitions.forEach { partition =>
-              val tp = new TopicPartition(topicName, partition.partitionIndex)
-              val apiError = Errors.forCode(partition.errorCode)
-              debug(s"Controller successfully handled AlterPartition request for $tp: $partition")
-              if (apiError == Errors.NONE) {
-                LeaderRecoveryState.optionalOf(partition.leaderRecoveryState).toScala match {
-                  case Some(leaderRecoveryState) =>
-                    partitionResponses(tp) = Right(
-                      new LeaderAndIsr(
-                        partition.leaderId,
-                        partition.leaderEpoch,
-                        partition.isr,
-                        leaderRecoveryState,
-                        partition.partitionEpoch
-                      )
+          topic.partitions.forEach { partition =>
+            val tp = new TopicIdPartition(topic.topicId, partition.partitionIndex)
+            val apiError = Errors.forCode(partition.errorCode)
+            debug(s"Controller successfully handled AlterPartition request for $tp: $partition")
+            if (apiError == Errors.NONE) {
+              LeaderRecoveryState.optionalOf(partition.leaderRecoveryState).toScala match {
+                case Some(leaderRecoveryState) =>
+                  partitionResponses(tp) = Right(
+                    new LeaderAndIsr(
+                      partition.leaderId,
+                      partition.leaderEpoch,
+                      partition.isr,
+                      leaderRecoveryState,
+                      partition.partitionEpoch
                     )
+                  )
 
-                  case None =>
-                    error(s"Controller returned an invalid leader recovery state (${partition.leaderRecoveryState}) for $tp: $partition")
-                    partitionResponses(tp) = Left(Errors.UNKNOWN_SERVER_ERROR)
-                }
-              } else {
-                partitionResponses(tp) = Left(apiError)
+                case None =>
+                  error(s"Controller returned an invalid leader recovery state (${partition.leaderRecoveryState}) for $tp: $partition")
+                  partitionResponses(tp) = Left(Errors.UNKNOWN_SERVER_ERROR)
               }
+            } else {
+              partitionResponses(tp) = Left(apiError)
             }
           }
         }
@@ -323,11 +289,11 @@ class DefaultAlterPartitionManager(
         // partition was somehow erroneously excluded from the response. Note that these callbacks are run from
         // the leaderIsrUpdateLock write lock in Partition#sendAlterPartitionRequest
         inflightAlterPartitionItems.foreach { inflightAlterPartition =>
-          partitionResponses.get(inflightAlterPartition.topicIdPartition.topicPartition) match {
+          partitionResponses.get(inflightAlterPartition.topicIdPartition) match {
             case Some(leaderAndIsrOrError) =>
               // Regardless of callback outcome, we need to clear from the unsent updates map to unblock further
               // updates. We clear it now to allow the callback to submit a new update if needed.
-              unsentIsrUpdates.remove(inflightAlterPartition.topicIdPartition.topicPartition)
+              unsentIsrUpdates.remove(inflightAlterPartition.topicIdPartition)
               leaderAndIsrOrError match {
                 case Left(error) => inflightAlterPartition.future.completeExceptionally(error.exception)
                 case Right(leaderAndIsr) => inflightAlterPartition.future.complete(leaderAndIsr)

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

@@ -87,7 +87,7 @@ class ControllerRegistrationManager(
   /**
    * The current metadata version that is in effect. Only read or written from the event queue thread.
    */
-  private var metadataVersion: MetadataVersion = MetadataVersion.MINIMUM_KRAFT_VERSION
+  private var metadataVersion: MetadataVersion = MetadataVersion.MINIMUM_VERSION
 
   /**
    * True if we're registered. Only read or written from the event queue thread.

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

@@ -182,7 +182,7 @@ object KafkaRaftServer {
     }
 
     // Load the BootstrapMetadata.
-    val bootstrapDirectory = new BootstrapDirectory(config.metadataLogDir, Optional.empty())
+    val bootstrapDirectory = new BootstrapDirectory(config.metadataLogDir)
     val bootstrapMetadata = bootstrapDirectory.read()
     (metaPropsEnsemble, bootstrapMetadata)
   }