From dc72786e97a6861bc39528a2be440e1f81f282c3 Mon Sep 17 00:00:00 2001 From: Justine Date: Wed, 17 Feb 2021 12:47:42 -0800 Subject: [PATCH 1/4] Changed becomeLeaderOrFollower to ignore partitions with invalid IDs and return UNKNOWN_TOPIC_ID error --- .../scala/kafka/server/ReplicaManager.scala | 94 ++++++++++--------- .../kafka/server/ReplicaManagerTest.scala | 59 +++++++++++- 2 files changed, 104 insertions(+), 49 deletions(-) diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index ba50c86dd4011..58d5734dbb7bd 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -1364,34 +1364,59 @@ class ReplicaManager(val config: KafkaConfig, Some(partition) } - // Next check partition's leader epoch + // Next check the topic ID and the partition's leader epoch partitionOpt.foreach { partition => val currentLeaderEpoch = partition.getLeaderEpoch val requestLeaderEpoch = partitionState.leaderEpoch - if (requestLeaderEpoch > currentLeaderEpoch) { - // If the leader epoch is valid record the epoch of the controller that made the leadership decision. - // This is useful while updating the isr to maintain the decision maker controller's epoch in the zookeeper path - if (partitionState.replicas.contains(localBrokerId)) - partitionStates.put(partition, partitionState) - else { - stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " + - s"correlation id $correlationId epoch $controllerEpoch for partition $topicPartition as itself is not " + - s"in assigned replica list ${partitionState.replicas.asScala.mkString(",")}") - responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION) + val id = topicIds.get(topicPartition.topic()) + var invalidId = false + + // Ensure we have not received a request from an older protocol + if (id != null && id != Uuid.ZERO_UUID) { + partition.log.foreach { log => + // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file. + // This is because if the broker previously wrote it to file, it would be recovered on restart after failure. + if (log.topicId == Uuid.ZERO_UUID) { + log.partitionMetadataFile.write(id) + log.topicId = id + // Warn if the topic ID in the request does not match the log. + } else if (log.topicId != id) { + stateChangeLogger.warn(s"Topic Id in memory: ${log.topicId.toString} does not" + + s" match the topic Id provided in the request: " + + s"${id.toString}.") + responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_ID) + invalidId = true + } + } + } + + // If we found an invalid ID, we don't need to check the leader epoch + if (!invalidId) { + if (requestLeaderEpoch > currentLeaderEpoch) { + // If the leader epoch is valid record the epoch of the controller that made the leadership decision. + // This is useful while updating the isr to maintain the decision maker controller's epoch in the zookeeper path + if (partitionState.replicas.contains(localBrokerId)) + partitionStates.put(partition, partitionState) + else { + stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " + + s"correlation id $correlationId epoch $controllerEpoch for partition $topicPartition as itself is not " + + s"in assigned replica list ${partitionState.replicas.asScala.mkString(",")}") + responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION) + } + } else if (requestLeaderEpoch < currentLeaderEpoch) { + stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " + + s"controller $controllerId with correlation id $correlationId " + + s"epoch $controllerEpoch for partition $topicPartition since its associated " + + s"leader epoch $requestLeaderEpoch is smaller than the current " + + s"leader epoch $currentLeaderEpoch") + responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) + } else { + stateChangeLogger.info(s"Ignoring LeaderAndIsr request from " + + s"controller $controllerId with correlation id $correlationId " + + s"epoch $controllerEpoch for partition $topicPartition since its associated " + + s"leader epoch $requestLeaderEpoch matches the current leader epoch") + responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) } - } else if (requestLeaderEpoch < currentLeaderEpoch) { - stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " + - s"controller $controllerId with correlation id $correlationId " + - s"epoch $controllerEpoch for partition $topicPartition since its associated " + - s"leader epoch $requestLeaderEpoch is smaller than the current " + - s"leader epoch $currentLeaderEpoch") - responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) - } else { - stateChangeLogger.info(s"Ignoring LeaderAndIsr request from " + - s"controller $controllerId with correlation id $correlationId " + - s"epoch $controllerEpoch for partition $topicPartition since its associated " + - s"leader epoch $requestLeaderEpoch matches the current leader epoch") - responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) } } } @@ -1424,27 +1449,8 @@ class ReplicaManager(val config: KafkaConfig, * In this case ReplicaManager.allPartitions will map this topic-partition to an empty Partition object. * we need to map this topic-partition to OfflinePartition instead. */ - val local = localLog(topicPartition) - if (local.isEmpty) + if (localLog(topicPartition).isEmpty) markPartitionOffline(topicPartition) - else { - val id = topicIds.get(topicPartition.topic()) - // Ensure we have not received a request from an older protocol - if (id != null && !id.equals(Uuid.ZERO_UUID)) { - val log = local.get - // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file. - // This is because if the broker previously wrote it to file, it would be recovered on restart after failure. - if (log.topicId.equals(Uuid.ZERO_UUID)) { - log.partitionMetadataFile.write(id) - log.topicId = id - // Warn if the topic ID in the request does not match the log. - } else if (!log.topicId.equals(id)) { - stateChangeLogger.warn(s"Topic Id in memory: ${log.topicId.toString} does not" + - s" match the topic Id provided in the request: " + - s"${id.toString}.") - } - } - } } // we initialize highwatermark thread after the first leaderisrrequest. This ensures that all the partitions diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index c31bf8efe54fe..80f1874321d06 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -2229,6 +2229,7 @@ class ReplicaManagerTest { .createLogIfNotExists(isNew = false, isFutureReplica = false, new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) val topicIds = Collections.singletonMap(topic, Uuid.randomUuid()) + val topicNames = topicIds.asScala.map(_.swap).asJava def leaderAndIsrRequest(epoch: Int): LeaderAndIsrRequest = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, Seq(new LeaderAndIsrPartitionState() @@ -2244,7 +2245,8 @@ class ReplicaManagerTest { topicIds, Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() - replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0), (_, _) => ()) + val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0), (_, _) => ()) + assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition)) assertFalse(replicaManager.localLog(topicPartition).isEmpty) val id = topicIds.get(topicPartition.topic()) val log = replicaManager.localLog(topicPartition).get @@ -2257,6 +2259,48 @@ class ReplicaManagerTest { } finally replicaManager.shutdown(checkpointHW = false) } + @Test + def testInvalidIdReturnsError() = { + val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time)) + try { + val brokerList = Seq[Integer](0, 1).asJava + val topicPartition = new TopicPartition(topic, 0) + replicaManager.createPartition(topicPartition) + .createLogIfNotExists(isNew = false, isFutureReplica = false, + new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) + val topicIds = Collections.singletonMap(topic, Uuid.randomUuid()) + val topicNames = topicIds.asScala.map(_.swap).asJava + + val invalidTopicIds = Collections.singletonMap(topic, Uuid.randomUuid()) + val invalidTopicNames = invalidTopicIds.asScala.map(_.swap).asJava + + def leaderAndIsrRequest(epoch: Int, topicIds: java.util.Map[String, Uuid]): LeaderAndIsrRequest = + new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion, 0, 0, brokerEpoch, + Seq(new LeaderAndIsrPartitionState() + .setTopicName(topic) + .setPartitionIndex(0) + .setControllerEpoch(0) + .setLeader(0) + .setLeaderEpoch(epoch) + .setIsr(brokerList) + .setZkVersion(0) + .setReplicas(brokerList) + .setIsNew(true)).asJava, + topicIds, + Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build() + + val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topicIds), (_, _) => ()) + assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition)) + + // Send request with invalid ID. + val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, invalidTopicIds), (_, _) => ()) + assertEquals(Errors.UNKNOWN_TOPIC_ID, response2.partitionErrors(invalidTopicNames).get(topicPartition)) + + val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, invalidTopicIds), (_, _) => ()) + assertEquals(Errors.UNKNOWN_TOPIC_ID, response3.partitionErrors(invalidTopicNames).get(topicPartition)) + } finally replicaManager.shutdown(checkpointHW = false) + } + @Test def testPartitionMetadataFileNotCreated() = { val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time)) @@ -2268,6 +2312,7 @@ class ReplicaManagerTest { .createLogIfNotExists(isNew = false, isFutureReplica = false, new LazyOffsetCheckpoints(replicaManager.highWatermarkCheckpoints)) val topicIds = Map(topic -> Uuid.ZERO_UUID, "foo" -> Uuid.randomUuid()).asJava + val topicNames = topicIds.asScala.map(_.swap).asJava def leaderAndIsrRequest(epoch: Int, name: String, version: Short): LeaderAndIsrRequest = LeaderAndIsrRequest.parse( new LeaderAndIsrRequest.Builder(version, 0, 0, brokerEpoch, @@ -2285,28 +2330,32 @@ class ReplicaManagerTest { Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build().serialize(), version) // There is no file if the topic does not have an associated topic ID. - replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "fakeTopic", ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ()) + val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "fakeTopic", ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ()) assertFalse(replicaManager.localLog(topicPartition).isEmpty) val log = replicaManager.localLog(topicPartition).get assertFalse(log.partitionMetadataFile.exists()) + assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition)) // There is no file if the topic has the default UUID. - replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topic, ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ()) + val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topic, ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ()) assertFalse(replicaManager.localLog(topicPartition).isEmpty) val log2 = replicaManager.localLog(topicPartition).get assertFalse(log2.partitionMetadataFile.exists()) + assertEquals(Errors.NONE, response2.partitionErrors(topicNames).get(topicPartition)) // There is no file if the request an older version - replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "foo", 0), (_, _) => ()) + val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "foo", 0), (_, _) => ()) assertFalse(replicaManager.localLog(topicPartitionFoo).isEmpty) val log3 = replicaManager.localLog(topicPartitionFoo).get assertFalse(log3.partitionMetadataFile.exists()) + assertEquals(Errors.NONE, response3.partitionErrors(topicNames).get(topicPartitionFoo)) // There is no file if the request is an older version - replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "foo", 4), (_, _) => ()) + val response4 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, "foo", 4), (_, _) => ()) assertFalse(replicaManager.localLog(topicPartitionFoo).isEmpty) val log4 = replicaManager.localLog(topicPartitionFoo).get assertFalse(log4.partitionMetadataFile.exists()) + assertEquals(Errors.NONE, response4.partitionErrors(topicNames).get(topicPartitionFoo)) } finally replicaManager.shutdown(checkpointHW = false) } From 89968d9690f82a0c75e86fcc0078503a70055fdb Mon Sep 17 00:00:00 2001 From: Justine Date: Wed, 17 Feb 2021 17:18:58 -0800 Subject: [PATCH 2/4] Addressed comments. Moved topic ID checks to Partition. --- .../errors/InconsistentTopicIdException.java | 27 +++++++ .../apache/kafka/common/protocol/Errors.java | 4 +- .../main/scala/kafka/cluster/Partition.scala | 31 ++++++- .../scala/kafka/server/ReplicaManager.scala | 81 ++++++++----------- .../kafka/server/ReplicaManagerTest.scala | 11 ++- 5 files changed, 99 insertions(+), 55 deletions(-) create mode 100644 clients/src/main/java/org/apache/kafka/common/errors/InconsistentTopicIdException.java diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InconsistentTopicIdException.java b/clients/src/main/java/org/apache/kafka/common/errors/InconsistentTopicIdException.java new file mode 100644 index 0000000000000..1dfe468564bf5 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/common/errors/InconsistentTopicIdException.java @@ -0,0 +1,27 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.errors; + +public class InconsistentTopicIdException extends InvalidMetadataException { + + private static final long serialVersionUID = 1L; + + public InconsistentTopicIdException(String message) { + super(message); + } + +} diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java index 03c1248055878..34c42064e78ea 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java @@ -47,6 +47,7 @@ import org.apache.kafka.common.errors.IllegalGenerationException; import org.apache.kafka.common.errors.IllegalSaslStateException; import org.apache.kafka.common.errors.InconsistentGroupProtocolException; +import org.apache.kafka.common.errors.InconsistentTopicIdException; import org.apache.kafka.common.errors.InconsistentVoterSetException; import org.apache.kafka.common.errors.InvalidCommitOffsetSizeException; import org.apache.kafka.common.errors.InvalidConfigurationException; @@ -354,7 +355,8 @@ public enum Errors { "Requested position is not greater than or equal to zero, and less than the size of the snapshot.", PositionOutOfRangeException::new), UNKNOWN_TOPIC_ID(100, "This server does not host this topic ID.", UnknownTopicIdException::new), - DUPLICATE_BROKER_REGISTRATION(101, "This broker ID is already in use.", DuplicateBrokerRegistrationException::new); + DUPLICATE_BROKER_REGISTRATION(101, "This broker ID is already in use.", DuplicateBrokerRegistrationException::new), + INCONSISTENT_TOPIC_ID(102, "The log's topic ID did not match the topic ID in the request", InconsistentTopicIdException::new); private static final Logger log = LoggerFactory.getLogger(Errors.class); diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index 6f6cb88e92230..b4907733b1b6a 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -40,7 +40,7 @@ import org.apache.kafka.common.record.{MemoryRecords, RecordBatch} import org.apache.kafka.common.requests._ import org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.{UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET} import org.apache.kafka.common.utils.Time -import org.apache.kafka.common.{IsolationLevel, TopicPartition} +import org.apache.kafka.common.{IsolationLevel, TopicPartition, Uuid} import scala.collection.{Map, Seq} import scala.jdk.CollectionConverters._ @@ -428,6 +428,35 @@ class Partition(val topicPartition: TopicPartition, this.log = Some(log) } + /** + * This method checks if the topic ID provided in the request is consistent with the topic ID in the log. + * If a valid topic ID is provided, but the log has no ID set, set the log ID to be the request ID. + * Returns a boolean representing whether the topic ID was consistent and the final log ID if it exists. + */ + def checkOrSetTopicId(requestTopicId: Uuid): (Boolean, Option[Uuid]) = { + // If the request had an invalid topic ID, then we assume that topic IDs are not supported. + // The topic ID was not inconsistent, so return true. + // If the log is empty, then we can not say that topic ID is inconsistent, so return true. + if (requestTopicId == null || requestTopicId == Uuid.ZERO_UUID) + (true, None) + else if (log.isEmpty) + (true, None) + else { + val partitionLog = log.get + // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file. + // This is because if the broker previously wrote it to file, it would be recovered on restart after failure. + // Topic ID is consistent since we are just setting it here. + if (partitionLog.topicId == Uuid.ZERO_UUID) { + partitionLog.partitionMetadataFile.write(requestTopicId) + partitionLog.topicId = requestTopicId + (true, Some(partitionLog.topicId)) + } else if (partitionLog.topicId != requestTopicId) + (false, Some(partitionLog.topicId)) + else + (true, Some(partitionLog.topicId)) + } + } + // remoteReplicas will be called in the hot path, and must be inexpensive def remoteReplicas: Iterable[Replica] = remoteReplicasMap.values diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 58d5734dbb7bd..40c38c05e4ffe 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -37,7 +37,7 @@ import kafka.server.metadata.ConfigRepository import kafka.utils._ import kafka.utils.Implicits._ import kafka.zk.KafkaZkClient -import org.apache.kafka.common.{ElectionType, IsolationLevel, Node, TopicPartition, Uuid} +import org.apache.kafka.common.{ElectionType, IsolationLevel, Node, TopicPartition} import org.apache.kafka.common.errors._ import org.apache.kafka.common.internals.Topic import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState @@ -1368,55 +1368,38 @@ class ReplicaManager(val config: KafkaConfig, partitionOpt.foreach { partition => val currentLeaderEpoch = partition.getLeaderEpoch val requestLeaderEpoch = partitionState.leaderEpoch - val id = topicIds.get(topicPartition.topic()) - var invalidId = false - - // Ensure we have not received a request from an older protocol - if (id != null && id != Uuid.ZERO_UUID) { - partition.log.foreach { log => - // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file. - // This is because if the broker previously wrote it to file, it would be recovered on restart after failure. - if (log.topicId == Uuid.ZERO_UUID) { - log.partitionMetadataFile.write(id) - log.topicId = id - // Warn if the topic ID in the request does not match the log. - } else if (log.topicId != id) { - stateChangeLogger.warn(s"Topic Id in memory: ${log.topicId.toString} does not" + - s" match the topic Id provided in the request: " + - s"${id.toString}.") - responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_ID) - invalidId = true - } - } - } - - // If we found an invalid ID, we don't need to check the leader epoch - if (!invalidId) { - if (requestLeaderEpoch > currentLeaderEpoch) { - // If the leader epoch is valid record the epoch of the controller that made the leadership decision. - // This is useful while updating the isr to maintain the decision maker controller's epoch in the zookeeper path - if (partitionState.replicas.contains(localBrokerId)) - partitionStates.put(partition, partitionState) - else { - stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " + - s"correlation id $correlationId epoch $controllerEpoch for partition $topicPartition as itself is not " + - s"in assigned replica list ${partitionState.replicas.asScala.mkString(",")}") - responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION) - } - } else if (requestLeaderEpoch < currentLeaderEpoch) { - stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " + - s"controller $controllerId with correlation id $correlationId " + - s"epoch $controllerEpoch for partition $topicPartition since its associated " + - s"leader epoch $requestLeaderEpoch is smaller than the current " + - s"leader epoch $currentLeaderEpoch") - responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) - } else { - stateChangeLogger.info(s"Ignoring LeaderAndIsr request from " + - s"controller $controllerId with correlation id $correlationId " + - s"epoch $controllerEpoch for partition $topicPartition since its associated " + - s"leader epoch $requestLeaderEpoch matches the current leader epoch") - responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) + val requestTopicId = topicIds.get(topicPartition.topic) + val (consistentTopicId, logTopicId) = partition.checkOrSetTopicId(requestTopicId) + + if (!consistentTopicId) { + stateChangeLogger.error(s"Topic Id in memory: ${logTopicId.get} does not" + + s" match the topic Id for partition $topicPartition provided in the request: " + + s"$requestTopicId.") + responseMap.put(topicPartition, Errors.INCONSISTENT_TOPIC_ID) + } else if (requestLeaderEpoch > currentLeaderEpoch) { + // If the leader epoch is valid record the epoch of the controller that made the leadership decision. + // This is useful while updating the isr to maintain the decision maker controller's epoch in the zookeeper path + if (partitionState.replicas.contains(localBrokerId)) + partitionStates.put(partition, partitionState) + else { + stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from controller $controllerId with " + + s"correlation id $correlationId epoch $controllerEpoch for partition $topicPartition as itself is not " + + s"in assigned replica list ${partitionState.replicas.asScala.mkString(",")}") + responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_OR_PARTITION) } + } else if (requestLeaderEpoch < currentLeaderEpoch) { + stateChangeLogger.warn(s"Ignoring LeaderAndIsr request from " + + s"controller $controllerId with correlation id $correlationId " + + s"epoch $controllerEpoch for partition $topicPartition since its associated " + + s"leader epoch $requestLeaderEpoch is smaller than the current " + + s"leader epoch $currentLeaderEpoch") + responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) + } else { + stateChangeLogger.info(s"Ignoring LeaderAndIsr request from " + + s"controller $controllerId with correlation id $correlationId " + + s"epoch $controllerEpoch for partition $topicPartition since its associated " + + s"leader epoch $requestLeaderEpoch matches the current leader epoch") + responseMap.put(topicPartition, Errors.STALE_CONTROLLER_EPOCH) } } } diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala index 80f1874321d06..9b289e578173d 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala @@ -2292,12 +2292,15 @@ class ReplicaManagerTest { val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topicIds), (_, _) => ()) assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition)) - // Send request with invalid ID. - val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, invalidTopicIds), (_, _) => ()) - assertEquals(Errors.UNKNOWN_TOPIC_ID, response2.partitionErrors(invalidTopicNames).get(topicPartition)) + val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, topicIds), (_, _) => ()) + assertEquals(Errors.NONE, response2.partitionErrors(topicNames).get(topicPartition)) + // Send request with invalid ID. val response3 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(1, invalidTopicIds), (_, _) => ()) - assertEquals(Errors.UNKNOWN_TOPIC_ID, response3.partitionErrors(invalidTopicNames).get(topicPartition)) + assertEquals(Errors.INCONSISTENT_TOPIC_ID, response3.partitionErrors(invalidTopicNames).get(topicPartition)) + + val response4 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(2, invalidTopicIds), (_, _) => ()) + assertEquals(Errors.INCONSISTENT_TOPIC_ID, response4.partitionErrors(invalidTopicNames).get(topicPartition)) } finally replicaManager.shutdown(checkpointHW = false) } From 06944ab1cc8cf6991b075be4fbb3dae457a86c0f Mon Sep 17 00:00:00 2001 From: Justine Date: Thu, 18 Feb 2021 09:30:16 -0800 Subject: [PATCH 3/4] Cleanups --- .../main/scala/kafka/cluster/Partition.scala | 44 +++++++++++-------- .../scala/kafka/server/ReplicaManager.scala | 6 +-- 2 files changed, 27 insertions(+), 23 deletions(-) diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index b4907733b1b6a..d9b782123dd82 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -430,30 +430,38 @@ class Partition(val topicPartition: TopicPartition, /** * This method checks if the topic ID provided in the request is consistent with the topic ID in the log. - * If a valid topic ID is provided, but the log has no ID set, set the log ID to be the request ID. - * Returns a boolean representing whether the topic ID was consistent and the final log ID if it exists. + * If a valid topic ID is provided, and the log exists but has no ID set, set the log ID to be the request ID. + * Returns a boolean representing whether the topic ID was consistent. */ - def checkOrSetTopicId(requestTopicId: Uuid): (Boolean, Option[Uuid]) = { + def checkOrSetTopicId(requestTopicId: Uuid): Boolean = { // If the request had an invalid topic ID, then we assume that topic IDs are not supported. // The topic ID was not inconsistent, so return true. // If the log is empty, then we can not say that topic ID is inconsistent, so return true. if (requestTopicId == null || requestTopicId == Uuid.ZERO_UUID) - (true, None) - else if (log.isEmpty) - (true, None) + true else { - val partitionLog = log.get - // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file. - // This is because if the broker previously wrote it to file, it would be recovered on restart after failure. - // Topic ID is consistent since we are just setting it here. - if (partitionLog.topicId == Uuid.ZERO_UUID) { - partitionLog.partitionMetadataFile.write(requestTopicId) - partitionLog.topicId = requestTopicId - (true, Some(partitionLog.topicId)) - } else if (partitionLog.topicId != requestTopicId) - (false, Some(partitionLog.topicId)) - else - (true, Some(partitionLog.topicId)) + log match { + case None => true + case Some(log) => { + // Check if topic ID is in memory, if not, it must be new to the broker and does not have a metadata file. + // This is because if the broker previously wrote it to file, it would be recovered on restart after failure. + // Topic ID is consistent since we are just setting it here. + if (log.topicId == Uuid.ZERO_UUID) { + log.partitionMetadataFile.write(requestTopicId) + log.topicId = requestTopicId + true + } else if (log.topicId != requestTopicId) { + // topic ID in log exists and is not consistent with request topic ID + stateChangeLogger.error(s"Topic Id in memory: ${log.topicId} does not" + + s" match the topic Id for partition $topicPartition provided in the request: " + + s"$requestTopicId.") + false + } else { + // topic ID in log exists and matches request topic ID + true + } + } + } } } diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala index 40c38c05e4ffe..820af7b6ece2c 100644 --- a/core/src/main/scala/kafka/server/ReplicaManager.scala +++ b/core/src/main/scala/kafka/server/ReplicaManager.scala @@ -1369,12 +1369,8 @@ class ReplicaManager(val config: KafkaConfig, val currentLeaderEpoch = partition.getLeaderEpoch val requestLeaderEpoch = partitionState.leaderEpoch val requestTopicId = topicIds.get(topicPartition.topic) - val (consistentTopicId, logTopicId) = partition.checkOrSetTopicId(requestTopicId) - if (!consistentTopicId) { - stateChangeLogger.error(s"Topic Id in memory: ${logTopicId.get} does not" + - s" match the topic Id for partition $topicPartition provided in the request: " + - s"$requestTopicId.") + if (!partition.checkOrSetTopicId(requestTopicId)) { responseMap.put(topicPartition, Errors.INCONSISTENT_TOPIC_ID) } else if (requestLeaderEpoch > currentLeaderEpoch) { // If the leader epoch is valid record the epoch of the controller that made the leadership decision. From 258d721918a77096b7473821c68cd4527aec5f2e Mon Sep 17 00:00:00 2001 From: Justine Date: Thu, 18 Feb 2021 10:32:51 -0800 Subject: [PATCH 4/4] Updated comment --- core/src/main/scala/kafka/cluster/Partition.scala | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala index d9b782123dd82..cfd029b38f55f 100755 --- a/core/src/main/scala/kafka/cluster/Partition.scala +++ b/core/src/main/scala/kafka/cluster/Partition.scala @@ -429,9 +429,11 @@ class Partition(val topicPartition: TopicPartition, } /** - * This method checks if the topic ID provided in the request is consistent with the topic ID in the log. + * Checks if the topic ID provided in the request is consistent with the topic ID in the log. * If a valid topic ID is provided, and the log exists but has no ID set, set the log ID to be the request ID. - * Returns a boolean representing whether the topic ID was consistent. + * + * @param requestTopicId the topic ID from the request + * @return true if the request topic id is consistent, false otherwise */ def checkOrSetTopicId(requestTopicId: Uuid): Boolean = { // If the request had an invalid topic ID, then we assume that topic IDs are not supported. @@ -451,7 +453,6 @@ class Partition(val topicPartition: TopicPartition, log.topicId = requestTopicId true } else if (log.topicId != requestTopicId) { - // topic ID in log exists and is not consistent with request topic ID stateChangeLogger.error(s"Topic Id in memory: ${log.topicId} does not" + s" match the topic Id for partition $topicPartition provided in the request: " + s"$requestTopicId.")