You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2018/08/28 00:39:59 UTC
[kafka] branch trunk updated: KAFKA-7347;
Return not leader error for OffsetsForLeaderEpoch requests to
non-replicas (#5576)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new ebd1354 KAFKA-7347; Return not leader error for OffsetsForLeaderEpoch requests to non-replicas (#5576)
ebd1354 is described below
commit ebd1354fd7578deb1cfb15846d7dbf8c9cf26074
Author: Jason Gustafson <ja...@confluent.io>
AuthorDate: Mon Aug 27 17:39:44 2018 -0700
KAFKA-7347; Return not leader error for OffsetsForLeaderEpoch requests to non-replicas (#5576)
The broker should return NOT_LEADER_FOR_PARTITION for OffsetsForLeaderEpoch requests against non-replicas instead of UNKNOWN_TOPIC_OR_PARTITION. This patch also fixes a minor bug in the handling of ListOffsets request using the DEBUG replica id. We should return UNKNOWN_TOPIC_OR_PARTITION if the topic doesn't exist.
Reviewers: Jun Rao <ju...@gmail.com>
---
.../main/scala/kafka/server/ReplicaManager.scala | 19 +++--
.../kafka/api/AdminClientIntegrationTest.scala | 2 +-
.../server/AlterReplicaLogDirsRequestTest.scala | 4 +-
.../unit/kafka/server/ListOffsetsRequestTest.scala | 88 ++++++++++++++++++++++
.../server/OffsetsForLeaderEpochRequestTest.scala | 65 ++++++++++++++++
.../server/epoch/LeaderEpochIntegrationTest.scala | 2 +-
6 files changed, 171 insertions(+), 9 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index 14e537e..59581e7 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -35,7 +35,6 @@ import org.apache.kafka.common.errors._
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.protocol.Errors.{KAFKA_STORAGE_ERROR, UNKNOWN_TOPIC_OR_PARTITION}
import org.apache.kafka.common.record._
import org.apache.kafka.common.requests.FetchResponse.AbortedTransaction
import org.apache.kafka.common.requests.DescribeLogDirsResponse.{LogDirInfo, ReplicaInfo}
@@ -405,12 +404,18 @@ class ReplicaManager(val config: KafkaConfig,
else
partition.getReplica(brokerId).getOrElse(
throw new ReplicaNotAvailableException(s"Replica $brokerId is not available for partition $topicPartition"))
- case None =>
+
+ case None if metadataCache.contains(topicPartition) =>
throw new ReplicaNotAvailableException(s"Replica $brokerId is not available for partition $topicPartition")
+
+ case None =>
+ throw new UnknownTopicOrPartitionException(s"Partition $topicPartition doesn't exist")
}
}
- def getReplicaOrException(topicPartition: TopicPartition): Replica = getReplicaOrException(topicPartition, localBrokerId)
+ def getReplicaOrException(topicPartition: TopicPartition): Replica = {
+ getReplicaOrException(topicPartition, localBrokerId)
+ }
def getLeaderReplicaIfLocal(topicPartition: TopicPartition): Replica = {
val (_, replica) = getPartitionAndLeaderReplicaIfLocal(topicPartition)
@@ -1475,11 +1480,15 @@ class ReplicaManager(val config: KafkaConfig,
val epochEndOffset = getPartition(tp) match {
case Some(partition) =>
if (partition eq ReplicaManager.OfflinePartition)
- new EpochEndOffset(KAFKA_STORAGE_ERROR, UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
+ new EpochEndOffset(Errors.KAFKA_STORAGE_ERROR, UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
else
partition.lastOffsetForLeaderEpoch(leaderEpoch)
+
+ case None if metadataCache.contains(tp) =>
+ new EpochEndOffset(Errors.NOT_LEADER_FOR_PARTITION, UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
+
case None =>
- new EpochEndOffset(UNKNOWN_TOPIC_OR_PARTITION, UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
+ new EpochEndOffset(Errors.UNKNOWN_TOPIC_OR_PARTITION, UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET)
}
tp -> epochEndOffset
}
diff --git a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
index 20cf038..672a669 100644
--- a/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
@@ -319,7 +319,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
new AlterReplicaLogDirsOptions).values.asScala.values
futures.foreach { future =>
val exception = intercept[ExecutionException](future.get)
- assertTrue(exception.getCause.isInstanceOf[ReplicaNotAvailableException])
+ assertTrue(exception.getCause.isInstanceOf[UnknownTopicOrPartitionException])
}
createTopic(topic, numPartitions = 1, replicationFactor = serverCount)
diff --git a/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala b/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
index e6fa1cb..e986805 100644
--- a/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AlterReplicaLogDirsRequestTest.scala
@@ -49,7 +49,7 @@ class AlterReplicaLogDirsRequestTest extends BaseRequestTest {
// The response should show error REPLICA_NOT_AVAILABLE for all partitions
(0 until partitionNum).foreach { partition =>
val tp = new TopicPartition(topic, partition)
- assertEquals(Errors.REPLICA_NOT_AVAILABLE, alterReplicaLogDirsResponse1.responses().get(tp))
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, alterReplicaLogDirsResponse1.responses().get(tp))
assertTrue(servers.head.logManager.getLog(tp).isEmpty)
}
@@ -85,7 +85,7 @@ class AlterReplicaLogDirsRequestTest extends BaseRequestTest {
partitionDirs1.put(new TopicPartition(topic, 1), validDir1)
val alterReplicaDirResponse1 = sendAlterReplicaLogDirsRequest(partitionDirs1.toMap)
assertEquals(Errors.LOG_DIR_NOT_FOUND, alterReplicaDirResponse1.responses().get(new TopicPartition(topic, 0)))
- assertEquals(Errors.REPLICA_NOT_AVAILABLE, alterReplicaDirResponse1.responses().get(new TopicPartition(topic, 1)))
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, alterReplicaDirResponse1.responses().get(new TopicPartition(topic, 1)))
createTopic(topic, 3, 1)
diff --git a/core/src/test/scala/unit/kafka/server/ListOffsetsRequestTest.scala b/core/src/test/scala/unit/kafka/server/ListOffsetsRequestTest.scala
new file mode 100644
index 0000000..6ee47ee
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/server/ListOffsetsRequestTest.scala
@@ -0,0 +1,88 @@
+/*
+ * 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 kafka.server
+
+import java.lang.{Long => JLong}
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.common.requests.{IsolationLevel, ListOffsetRequest, ListOffsetResponse}
+import org.junit.Assert._
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class ListOffsetsRequestTest extends BaseRequestTest {
+
+ @Test
+ def testListOffsetsErrorCodes(): Unit = {
+ val topic = "topic"
+ val partition = new TopicPartition(topic, 0)
+
+ val consumerRequest = ListOffsetRequest.Builder
+ .forConsumer(false, IsolationLevel.READ_UNCOMMITTED)
+ .setTargetTimes(Map(partition -> ListOffsetRequest.EARLIEST_TIMESTAMP.asInstanceOf[JLong]).asJava)
+ .build()
+
+ val replicaRequest = ListOffsetRequest.Builder
+ .forReplica(ApiKeys.LIST_OFFSETS.latestVersion, servers.head.config.brokerId)
+ .setTargetTimes(Map(partition -> ListOffsetRequest.EARLIEST_TIMESTAMP.asInstanceOf[JLong]).asJava)
+ .build()
+
+ val debugReplicaRequest = ListOffsetRequest.Builder
+ .forReplica(ApiKeys.LIST_OFFSETS.latestVersion, ListOffsetRequest.DEBUGGING_REPLICA_ID)
+ .setTargetTimes(Map(partition -> ListOffsetRequest.EARLIEST_TIMESTAMP.asInstanceOf[JLong]).asJava)
+ .build()
+
+ // Unknown topic
+ val randomBrokerId = servers.head.config.brokerId
+ assertResponseError(Errors.UNKNOWN_TOPIC_OR_PARTITION, randomBrokerId, consumerRequest)
+ assertResponseError(Errors.UNKNOWN_TOPIC_OR_PARTITION, randomBrokerId, replicaRequest)
+ assertResponseError(Errors.UNKNOWN_TOPIC_OR_PARTITION, randomBrokerId, debugReplicaRequest)
+
+ val partitionToLeader = TestUtils.createTopic(zkClient, topic, numPartitions = 1, replicationFactor = 2, servers)
+ val replicas = zkClient.getReplicasForPartition(partition).toSet
+ val leader = partitionToLeader(partition.partition)
+ val follower = replicas.find(_ != leader).get
+ val nonReplica = servers.map(_.config.brokerId).find(!replicas.contains(_)).get
+
+ // Follower
+ assertResponseError(Errors.NOT_LEADER_FOR_PARTITION, follower, consumerRequest)
+ assertResponseError(Errors.NOT_LEADER_FOR_PARTITION, follower, replicaRequest)
+ assertResponseError(Errors.NONE, follower, debugReplicaRequest)
+
+ // Non-replica
+ assertResponseError(Errors.NOT_LEADER_FOR_PARTITION, nonReplica, consumerRequest)
+ assertResponseError(Errors.NOT_LEADER_FOR_PARTITION, nonReplica, replicaRequest)
+ assertResponseError(Errors.REPLICA_NOT_AVAILABLE, nonReplica, debugReplicaRequest)
+ }
+
+ private def assertResponseError(error: Errors, brokerId: Int, request: ListOffsetRequest): Unit = {
+ val response = sendRequest(brokerId, request)
+ assertEquals(request.partitionTimestamps.size, response.responseData.size)
+ response.responseData.asScala.values.foreach { partitionData =>
+ assertEquals(error, partitionData.error)
+ }
+ }
+
+ private def sendRequest(leaderId: Int, request: ListOffsetRequest): ListOffsetResponse = {
+ val response = connectAndSend(request, ApiKeys.LIST_OFFSETS, destination = brokerSocketServer(leaderId))
+ ListOffsetResponse.parse(response, request.version)
+ }
+
+}
diff --git a/core/src/test/scala/unit/kafka/server/OffsetsForLeaderEpochRequestTest.scala b/core/src/test/scala/unit/kafka/server/OffsetsForLeaderEpochRequestTest.scala
new file mode 100644
index 0000000..c6385f3
--- /dev/null
+++ b/core/src/test/scala/unit/kafka/server/OffsetsForLeaderEpochRequestTest.scala
@@ -0,0 +1,65 @@
+/*
+ * 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 kafka.server
+
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.common.requests.{OffsetsForLeaderEpochRequest, OffsetsForLeaderEpochResponse}
+import org.junit.Assert._
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class OffsetsForLeaderEpochRequestTest extends BaseRequestTest {
+
+ @Test
+ def testOffsetsForLeaderEpochErrorCodes(): Unit = {
+ val topic = "topic"
+ val partition = new TopicPartition(topic, 0)
+
+ val request = new OffsetsForLeaderEpochRequest.Builder(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion)
+ .add(partition, 0)
+ .build()
+
+ // Unknown topic
+ val randomBrokerId = servers.head.config.brokerId
+ assertResponseError(Errors.UNKNOWN_TOPIC_OR_PARTITION, randomBrokerId, request)
+
+ val partitionToLeader = TestUtils.createTopic(zkClient, topic, numPartitions = 1, replicationFactor = 2, servers)
+ val replicas = zkClient.getReplicasForPartition(partition).toSet
+ val leader = partitionToLeader(partition.partition)
+ val follower = replicas.find(_ != leader).get
+ val nonReplica = servers.map(_.config.brokerId).find(!replicas.contains(_)).get
+
+ assertResponseError(Errors.NOT_LEADER_FOR_PARTITION, follower, request)
+ assertResponseError(Errors.NOT_LEADER_FOR_PARTITION, nonReplica, request)
+ }
+
+ private def assertResponseError(error: Errors, brokerId: Int, request: OffsetsForLeaderEpochRequest): Unit = {
+ val response = sendRequest(brokerId, request)
+ assertEquals(request.epochsByTopicPartition.size, response.responses.size)
+ response.responses.asScala.values.foreach { partitionData =>
+ assertEquals(error, partitionData.error)
+ }
+ }
+
+ private def sendRequest(leaderId: Int, request: OffsetsForLeaderEpochRequest): OffsetsForLeaderEpochResponse = {
+ val response = connectAndSend(request, ApiKeys.OFFSET_FOR_LEADER_EPOCH, destination = brokerSocketServer(leaderId))
+ OffsetsForLeaderEpochResponse.parse(response, request.version)
+ }
+}
diff --git a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala
index a6b7732..17683f4 100644
--- a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala
@@ -128,7 +128,7 @@ class LeaderEpochIntegrationTest extends ZooKeeperTestHarness with Logging {
//And should get no leader for partition error from t1p1 (as it's not on broker 0)
assertTrue(offsetsForEpochs(t1p1).hasError)
- assertEquals(UNKNOWN_TOPIC_OR_PARTITION, offsetsForEpochs(t1p1).error)
+ assertEquals(NOT_LEADER_FOR_PARTITION, offsetsForEpochs(t1p1).error)
assertEquals(UNDEFINED_EPOCH_OFFSET, offsetsForEpochs(t1p1).endOffset)
//Repointing to broker 1 we should get the correct offset for t1p1