7 Star 5 Fork 24

src-openEuler/kafka

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
0013-AlterIsr.patch 11.11 KB
一键复制 编辑 原始数据 按行查看 历史
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index a58f4238ff..88b337311d 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -581,9 +581,6 @@ class Partition(val topicPartition: TopicPartition,
leaderEpochStartOffsetOpt = Some(leaderEpochStartOffset)
zkVersion = partitionState.zkVersion
- // Clear any pending AlterIsr requests and check replica state
- alterIsrManager.clearPending(topicPartition)
-
// In the case of successive leader elections in a short time period, a follower may have
// entries in its log from a later epoch than any entry in the new leader's log. In order
// to ensure that these followers can truncate to the right offset, we must cache the new
@@ -661,9 +658,6 @@ class Partition(val topicPartition: TopicPartition,
leaderEpochStartOffsetOpt = None
zkVersion = partitionState.zkVersion
- // Since we might have been a leader previously, still clear any pending AlterIsr requests
- alterIsrManager.clearPending(topicPartition)
-
if (leaderReplicaIdOpt.contains(newLeaderBrokerId) && leaderEpoch == oldLeaderEpoch) {
false
} else {
@@ -1373,13 +1367,15 @@ class Partition(val topicPartition: TopicPartition,
isrState = proposedIsrState
if (!alterIsrManager.submit(alterIsrItem)) {
- // If the ISR manager did not accept our update, we need to revert back to previous state
+ // If the ISR manager did not accept our update, we need to revert the proposed state.
+ // This can happen if the ISR state was updated by the controller (via LeaderAndIsr in ZK-mode or
+ // ChangePartitionRecord in KRaft mode) but we have an AlterIsr request still in-flight.
isrState = oldState
isrChangeListener.markFailed()
- throw new IllegalStateException(s"Failed to enqueue ISR change state $newLeaderAndIsr for partition $topicPartition")
+ warn(s"Failed to enqueue ISR change state $newLeaderAndIsr for partition $topicPartition")
+ } else {
+ debug(s"Enqueued ISR change to state $newLeaderAndIsr after transition to $proposedIsrState")
}
-
- debug(s"Enqueued ISR change to state $newLeaderAndIsr after transition to $proposedIsrState")
}
/**
diff --git a/core/src/main/scala/kafka/server/AlterIsrManager.scala b/core/src/main/scala/kafka/server/AlterIsrManager.scala
index 9ad734f708..1059a3df3e 100644
--- a/core/src/main/scala/kafka/server/AlterIsrManager.scala
+++ b/core/src/main/scala/kafka/server/AlterIsrManager.scala
@@ -49,8 +49,6 @@ trait AlterIsrManager {
def shutdown(): Unit = {}
def submit(alterIsrItem: AlterIsrItem): Boolean
-
- def clearPending(topicPartition: TopicPartition): Unit
}
case class AlterIsrItem(topicPartition: TopicPartition,
@@ -134,9 +132,6 @@ class DefaultAlterIsrManager(
enqueued
}
- override def clearPending(topicPartition: TopicPartition): Unit = {
- unsentIsrUpdates.remove(topicPartition)
- }
private[server] def maybePropagateIsrChanges(): Unit = {
// Send all pending items if there is not already a request in-flight.
diff --git a/core/src/main/scala/kafka/server/ZkIsrManager.scala b/core/src/main/scala/kafka/server/ZkIsrManager.scala
index 2d88aac6b4..8dffcdf307 100644
--- a/core/src/main/scala/kafka/server/ZkIsrManager.scala
+++ b/core/src/main/scala/kafka/server/ZkIsrManager.scala
@@ -55,12 +55,6 @@ class ZkIsrManager(scheduler: Scheduler, time: Time, zkClient: KafkaZkClient) ex
period = isrChangeNotificationConfig.checkIntervalMs, unit = TimeUnit.MILLISECONDS)
}
- override def clearPending(topicPartition: TopicPartition): Unit = {
- // Since we always immediately process ZK updates and never actually enqueue anything, there is nothing to
- // clear here so this is a no-op. Even if there are changes that have not been propagated, the write to ZK
- // has already happened, so we may as well send the notification to the controller.
- }
-
override def submit(alterIsrItem: AlterIsrItem): Boolean = {
debug(s"Writing new ISR ${alterIsrItem.leaderAndIsr.isr} to ZooKeeper with version " +
s"${alterIsrItem.leaderAndIsr.zkVersion} for partition ${alterIsrItem.topicPartition}")
diff --git a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
index 5eedb63ae5..4dbd735753 100644
--- a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala
@@ -18,10 +18,10 @@
package kafka.controller
import java.util.Properties
-import java.util.concurrent.{CountDownLatch, LinkedBlockingQueue}
-
+import java.util.concurrent.{CompletableFuture, CountDownLatch, LinkedBlockingQueue, TimeUnit}
import com.yammer.metrics.core.Timer
import kafka.api.{ApiVersion, KAFKA_2_6_IV0, KAFKA_2_7_IV0, LeaderAndIsr}
+import kafka.controller.KafkaController.AlterIsrCallback
import kafka.metrics.KafkaYammerMetrics
import kafka.server.{KafkaConfig, KafkaServer}
import kafka.utils.{LogCaptureAppender, TestUtils}
@@ -849,6 +849,67 @@ class ControllerIntegrationTest extends ZooKeeperTestHarness {
latch.await()
}
+ @Test
+ def testAlterIsrErrors(): Unit = {
+ servers = makeServers(1)
+ val controllerId = TestUtils.waitUntilControllerElected(zkClient)
+ val tp = new TopicPartition("t", 0)
+ val assignment = Map(tp.partition -> Seq(controllerId))
+ TestUtils.createTopic(zkClient, tp.topic, partitionReplicaAssignment = assignment, servers = servers)
+ val controller = getController().kafkaController
+ var future = captureAlterIsrError(controllerId, controller.brokerEpoch - 1,
+ Map(tp -> LeaderAndIsr(controllerId, List(controllerId))))
+ var capturedError = future.get(5, TimeUnit.SECONDS)
+ assertEquals(Errors.STALE_BROKER_EPOCH, capturedError)
+
+ future = captureAlterIsrError(99, controller.brokerEpoch,
+ Map(tp -> LeaderAndIsr(controllerId, List(controllerId))))
+ capturedError = future.get(5, TimeUnit.SECONDS)
+ assertEquals(Errors.STALE_BROKER_EPOCH, capturedError)
+
+ val unknownTopicPartition = new TopicPartition("unknown", 99)
+ future = captureAlterIsrPartitionError(controllerId, controller.brokerEpoch,
+ Map(unknownTopicPartition -> LeaderAndIsr(controllerId, List(controllerId))), unknownTopicPartition)
+ capturedError = future.get(5, TimeUnit.SECONDS)
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, capturedError)
+
+ future = captureAlterIsrPartitionError(controllerId, controller.brokerEpoch,
+ Map(tp -> LeaderAndIsr(controllerId, 1, List(controllerId), 99)), tp)
+ capturedError = future.get(5, TimeUnit.SECONDS)
+ assertEquals(Errors.INVALID_UPDATE_VERSION, capturedError)
+ }
+
+ def captureAlterIsrError(brokerId: Int, brokerEpoch: Long, isrsToAlter: Map[TopicPartition, LeaderAndIsr]): CompletableFuture[Errors] = {
+ val future = new CompletableFuture[Errors]()
+ val controller = getController().kafkaController
+ val callback: AlterIsrCallback = {
+ case Left(_: Map[TopicPartition, Either[Errors, LeaderAndIsr]]) =>
+ future.completeExceptionally(new AssertionError(s"Should have seen top-level error"))
+ case Right(error: Errors) =>
+ future.complete(error)
+ }
+ controller.eventManager.put(AlterIsrReceived(brokerId, brokerEpoch, isrsToAlter, callback))
+ future
+ }
+
+ def captureAlterIsrPartitionError(brokerId: Int, brokerEpoch: Long, isrsToAlter: Map[TopicPartition, LeaderAndIsr], tp: TopicPartition): CompletableFuture[Errors] = {
+ val future = new CompletableFuture[Errors]()
+ val controller = getController().kafkaController
+ val callback: AlterIsrCallback = {
+ case Left(partitionResults: Map[TopicPartition, Either[Errors, LeaderAndIsr]]) =>
+ partitionResults.get(tp) match {
+ case Some(Left(error: Errors)) => future.complete(error)
+ case Some(Right(_: LeaderAndIsr)) => future.completeExceptionally(new AssertionError(s"Should have seen an error for $tp in result"))
+ case None => future.completeExceptionally(new AssertionError(s"Should have seen $tp in result"))
+ }
+ case Right(_: Errors) =>
+ future.completeExceptionally(new AssertionError(s"Should not seen top-level error"))
+ }
+ controller.eventManager.put(AlterIsrReceived(brokerId, brokerEpoch, isrsToAlter, callback))
+ future
+ }
+
+
@Test
def testTopicIdsAreAdded(): Unit = {
servers = makeServers(1)
diff --git a/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala b/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala
index 1074fd3157..1c8c81471f 100644
--- a/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AlterIsrManagerTest.scala
@@ -70,8 +70,10 @@ class AlterIsrManagerTest {
@Test
def testOverwriteWithinBatch(): Unit = {
val capture = EasyMock.newCapture[AbstractRequest.Builder[AlterIsrRequest]]()
+ val callbackCapture = EasyMock.newCapture[ControllerRequestCompletionHandler]()
+
EasyMock.expect(brokerToController.start())
- EasyMock.expect(brokerToController.sendRequest(EasyMock.capture(capture), EasyMock.anyObject())).once()
+ EasyMock.expect(brokerToController.sendRequest(EasyMock.capture(capture), EasyMock.capture(callbackCapture))).times(2)
EasyMock.replay(brokerToController)
val scheduler = new MockScheduler(time)
@@ -81,11 +83,21 @@ class AlterIsrManagerTest {
// Only send one ISR update for a given topic+partition
assertTrue(alterIsrManager.submit(AlterIsrItem(tp0, new LeaderAndIsr(1, 1, List(1,2,3), 10), _ => {}, 0)))
assertFalse(alterIsrManager.submit(AlterIsrItem(tp0, new LeaderAndIsr(1, 1, List(1,2), 10), _ => {}, 0)))
+
+ // Simulate response
+ val alterIsrResp = partitionResponse(tp0, Errors.NONE)
+ val resp = new ClientResponse(null, null, "", 0L, 0L,
+ false, null, null, alterIsrResp)
+ callbackCapture.getValue.onComplete(resp)
+
+ // Now we can submit this partition again
+ assertTrue(alterIsrManager.submit(AlterIsrItem(tp0, new LeaderAndIsr(1, 1, List(1), 10), _ => {}, 0)))
EasyMock.verify(brokerToController)
+ // Make sure we sent the right request ISR={1}
val request = capture.getValue.build()
assertEquals(request.data().topics().size(), 1)
- assertEquals(request.data().topics().get(0).partitions().get(0).newIsr().size(), 3)
+ assertEquals(request.data().topics().get(0).partitions().get(0).newIsr().size(), 1)
}
@Test
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 43df2b97f4..8e52007bc7 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -1106,10 +1106,6 @@ object TestUtils extends Logging {
}
}
- override def clearPending(topicPartition: TopicPartition): Unit = {
- inFlight.set(false);
- }
-
def completeIsrUpdate(newZkVersion: Int): Unit = {
if (inFlight.compareAndSet(true, false)) {
val item = isrUpdates.head
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/src-openeuler/kafka.git
[email protected]:src-openeuler/kafka.git
src-openeuler
kafka
kafka
master

搜索帮助