You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/08/09 14:57:45 UTC
[GitHub] [kafka] jsancio opened a new pull request #11189: KAFKA-13161: Always update the partition state
jsancio opened a new pull request #11189:
URL: https://github.com/apache/kafka/pull/11189
When processing the topics delta, always update the partition state when
transitioning a replica to follower.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685417821
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2204,32 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
- // Create the local replica even if the leader is unavailable. This is required
- // to ensure that we include the partition's high watermark in the checkpoint
- // file (see KAFKA-1647)
- partition.createLogIfNotExists(isNew, false, offsetCheckpoints, Some(info.topicId))
-
if (shuttingDown) {
stateChangeLogger.trace(s"Unable to start fetching ${tp} with topic " +
s"ID ${info.topicId} because the replica manager is shutting down.")
} else {
val listenerName = config.interBrokerListenerName.value()
val leader = info.partition.leader
- Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
- case None => stateChangeLogger.trace(s"Unable to start fetching ${tp} " +
- s"with topic ID ${info.topicId} from leader ${leader} because it is not " +
- "alive.")
- case Some(node) =>
- val leaderEndPoint = new BrokerEndPoint(node.id(), node.host(), node.port())
- val log = partition.localLogOrException
- val fetchOffset = initialFetchOffset(log)
- partitionsToMakeFollower.put(tp,
- InitialFetchState(leaderEndPoint, partition.getLeaderEpoch, fetchOffset))
+ val state = info.partition.toLeaderAndIsrPartitionState(tp, isNew)
+ if (partition.makeFollower(state, offsetCheckpoints, Some(info.topicId))) {
+ Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
Review comment:
The order here is a little different than in the LeaderAndIsr path. In the other path, we skip the call to `makeFollower` if the leader is not available. I think that implicitly also handles the case when the current leader is -1. I noticed that `Partition.makeFollower` does not check for this case when it does the assignment:
```scala
leaderReplicaIdOpt = Some(newLeaderBrokerId)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685611958
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2203,37 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
Review comment:
Hmm, as I'm looking over the code, this seems wrong. We'd want to trigger delayed operations after the call to `makeFollower` since the change from leader to follower is what would complete the operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685608200
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -2810,4 +2814,183 @@ class ReplicaManagerTest {
Replicas.NONE, Replicas.NONE, 2, 123, 456)))),
replicaManager.calculateDeltaChanges(TEST_DELTA))
}
+
+ @Test
+ def testDeltaFromLeaderToFollower(): Unit = {
+ val localId = 1
+ val otherId = localId + 1
+ val numOfRecords = 3
+ val epoch = 100
+ val topicPartition = new TopicPartition("foo", 0)
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), localId)
+
+ // Make the local replica the leader
+ val leaderMetadataImage = imageFromTopics(topicsImage(localId, true, epoch))
+ replicaManager.applyDelta(leaderMetadataImage, topicsDelta(localId, true, epoch))
+
+ // Check the state of that partition and fetcher
+ val HostedPartition.Online(leaderPartition) = replicaManager.getPartition(topicPartition)
+ assertTrue(leaderPartition.isLeader)
+ assertEquals(Set(localId, otherId), leaderPartition.inSyncReplicaIds)
+ assertEquals(epoch, leaderPartition.getLeaderEpoch)
+
+ assertEquals(None, replicaManager.replicaFetcherManager.getFetcher(topicPartition))
+
+ // Send a produce request and advance the highwatermark
+ val leaderResponse = sendProducerAppend(replicaManager, topicPartition, numOfRecords)
+ fetchMessages(
+ replicaManager,
+ otherId,
+ topicPartition,
+ new PartitionData(numOfRecords, 0, Int.MaxValue, Optional.empty()),
+ Int.MaxValue,
+ IsolationLevel.READ_UNCOMMITTED,
+ None
+ )
+ assertEquals(Errors.NONE, leaderResponse.get.error)
+
+ // Change the local replica to follower
+ val followerMetadataImage = imageFromTopics(topicsImage(localId, false, epoch + 1))
+ replicaManager.applyDelta(followerMetadataImage, topicsDelta(localId, false, epoch + 1))
+
+ // Append on a follower should fail
+ val followerResponse = sendProducerAppend(replicaManager, topicPartition, numOfRecords)
+ assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, followerResponse.get.error)
+
+ // Check the state of that partition and fetcher
+ val HostedPartition.Online(followerPartition) = replicaManager.getPartition(topicPartition)
+ assertFalse(followerPartition.isLeader)
+ assertEquals(epoch + 1, followerPartition.getLeaderEpoch)
+
+ val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
+ assertNotEquals(None, fetcher)
+ assertEquals(BrokerEndPoint(otherId, "localhost", 9093), fetcher.get.sourceBroker)
+ }
+
+ @Test
+ def testDeltaFromFollowerToLeader(): Unit = {
+ val localId = 1
+ val otherId = localId + 1
+ val numOfRecords = 3
+ val epoch = 100
+ val topicPartition = new TopicPartition("foo", 0)
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), localId)
+
+ // Make the local replica the follower
+ val followerMetadataImage = imageFromTopics(topicsImage(localId, false, epoch))
+ replicaManager.applyDelta(followerMetadataImage, topicsDelta(localId, false, epoch))
+
+ // Check the state of that partition and fetcher
+ val HostedPartition.Online(followerPartition) = replicaManager.getPartition(topicPartition)
+ assertFalse(followerPartition.isLeader)
+ assertEquals(epoch, followerPartition.getLeaderEpoch)
+
+ val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
+ assertNotEquals(None, fetcher)
+ assertEquals(BrokerEndPoint(otherId, "localhost", 9093), fetcher.get.sourceBroker)
+
+ // Append on a follower should fail
+ val followerResponse = sendProducerAppend(replicaManager, topicPartition, numOfRecords)
+ assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, followerResponse.get.error)
+
+ // Change the local replica to leader
+ val leaderMetadataImage = imageFromTopics(topicsImage(localId, true, epoch + 1))
+ replicaManager.applyDelta(leaderMetadataImage, topicsDelta(localId, true, epoch + 1))
+
+ // Send a produce request and advance the highwatermark
+ val leaderResponse = sendProducerAppend(replicaManager, topicPartition, numOfRecords)
+ fetchMessages(
+ replicaManager,
+ otherId,
+ topicPartition,
+ new PartitionData(numOfRecords, 0, Int.MaxValue, Optional.empty()),
+ Int.MaxValue,
+ IsolationLevel.READ_UNCOMMITTED,
+ None
+ )
+ assertEquals(Errors.NONE, leaderResponse.get.error)
+
+ val HostedPartition.Online(leaderPartition) = replicaManager.getPartition(topicPartition)
+ assertTrue(leaderPartition.isLeader)
+ assertEquals(Set(localId, otherId), leaderPartition.inSyncReplicaIds)
+ assertEquals(epoch + 1, leaderPartition.getLeaderEpoch)
+
+ assertEquals(None, replicaManager.replicaFetcherManager.getFetcher(topicPartition))
+ }
+
+ @Test
+ def testDeltaFollowerWithNoChange(): Unit = {
+ val localId = 1
+ val otherId = localId + 1
+ val epoch = 100
+ val topicPartition = new TopicPartition("foo", 0)
+ val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time), localId)
+
+ // Make the local replica the follower
+ val followerMetadataImage = imageFromTopics(topicsImage(localId, false, epoch))
+ replicaManager.applyDelta(followerMetadataImage, topicsDelta(localId, false, epoch))
+
+ // Check the state of that partition and fetcher
+ val HostedPartition.Online(followerPartition) = replicaManager.getPartition(topicPartition)
+ assertFalse(followerPartition.isLeader)
+ assertEquals(epoch, followerPartition.getLeaderEpoch)
+
+ val fetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
+ assertNotEquals(None, fetcher)
+ assertEquals(BrokerEndPoint(otherId, "localhost", 9093), fetcher.get.sourceBroker)
+
+ // Apply the same delta again
+ replicaManager.applyDelta(followerMetadataImage, topicsDelta(localId, false, epoch))
+
+ // Check that the state stays the same
+ val HostedPartition.Online(noChangePartition) = replicaManager.getPartition(topicPartition)
+ assertFalse(noChangePartition.isLeader)
+ assertEquals(epoch, noChangePartition.getLeaderEpoch)
+
+ val noChangeFetcher = replicaManager.replicaFetcherManager.getFetcher(topicPartition)
+ assertNotEquals(None, noChangeFetcher)
+ assertEquals(BrokerEndPoint(otherId, "localhost", 9093), noChangeFetcher.get.sourceBroker)
Review comment:
nit: maybe a simple way to write these two assertions:
```
assertEquals(Some(BrokerEndPoint(otherId, "localhost", 9093)), noChangeFetcher.map(_.sourceBroker))
```
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2235,8 +2244,15 @@ class ReplicaManager(val config: KafkaConfig,
}
}
}
- updateLeaderAndFollowerMetrics(newFollowerTopicSet)
+
+ replicaFetcherManager.removeFetcherForPartitions(partitionsToMakeFollower.keySet)
+ stateChangeLogger.info(s"Stopped fetchers as part of become-follower for ${partitionsToMakeFollower.size} partitions")
+
replicaFetcherManager.addFetcherForPartitions(partitionsToMakeFollower)
+ stateChangeLogger.info(s"Started fetchers as part of become-follower for ${partitionsToMakeFollower.size} partitions")
+
+ updateLeaderAndFollowerMetrics(newFollowerTopicSet)
+
Review comment:
nit: remove unneeded newline
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11189: KAFKA-13161: Update replica partition state and replica fetcher state on follower update
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r686152042
##########
File path: core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
##########
@@ -1084,40 +1090,46 @@ class ReplicaManagerTest {
topicPartition, leaderEpoch + leaderEpochIncrement, followerBrokerId,
leaderBrokerId, countDownLatch, expectTruncation = true, topicId = Some(topicId))
- val brokerList = Seq[Integer](0, 1).asJava
+ try {
Review comment:
Most of the changes here are indentation changes to add `try { ... } finally { replicaManager.shutdown() }`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685582747
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2203,32 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
- // Create the local replica even if the leader is unavailable. This is required
Review comment:
We seem to have dropped this logic. Do we not still need it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685440981
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2204,32 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
- // Create the local replica even if the leader is unavailable. This is required
- // to ensure that we include the partition's high watermark in the checkpoint
- // file (see KAFKA-1647)
- partition.createLogIfNotExists(isNew, false, offsetCheckpoints, Some(info.topicId))
-
if (shuttingDown) {
stateChangeLogger.trace(s"Unable to start fetching ${tp} with topic " +
s"ID ${info.topicId} because the replica manager is shutting down.")
} else {
val listenerName = config.interBrokerListenerName.value()
val leader = info.partition.leader
- Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
- case None => stateChangeLogger.trace(s"Unable to start fetching ${tp} " +
- s"with topic ID ${info.topicId} from leader ${leader} because it is not " +
- "alive.")
- case Some(node) =>
- val leaderEndPoint = new BrokerEndPoint(node.id(), node.host(), node.port())
- val log = partition.localLogOrException
- val fetchOffset = initialFetchOffset(log)
- partitionsToMakeFollower.put(tp,
- InitialFetchState(leaderEndPoint, partition.getLeaderEpoch, fetchOffset))
+ val state = info.partition.toLeaderAndIsrPartitionState(tp, isNew)
+ if (partition.makeFollower(state, offsetCheckpoints, Some(info.topicId))) {
+ Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
Review comment:
Okay. Let me fix that. I think we need to distinguish between no leader and a leader but no endpoint information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685590380
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2203,32 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
- // Create the local replica even if the leader is unavailable. This is required
Review comment:
Yes. Good catch. To match the ZK implementation, we need to at least create the log when partition.makeFollower is not called.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685271449
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2204,32 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
- // Create the local replica even if the leader is unavailable. This is required
- // to ensure that we include the partition's high watermark in the checkpoint
- // file (see KAFKA-1647)
- partition.createLogIfNotExists(isNew, false, offsetCheckpoints, Some(info.topicId))
-
if (shuttingDown) {
stateChangeLogger.trace(s"Unable to start fetching ${tp} with topic " +
s"ID ${info.topicId} because the replica manager is shutting down.")
} else {
val listenerName = config.interBrokerListenerName.value()
val leader = info.partition.leader
- Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
- case None => stateChangeLogger.trace(s"Unable to start fetching ${tp} " +
- s"with topic ID ${info.topicId} from leader ${leader} because it is not " +
- "alive.")
- case Some(node) =>
- val leaderEndPoint = new BrokerEndPoint(node.id(), node.host(), node.port())
- val log = partition.localLogOrException
- val fetchOffset = initialFetchOffset(log)
- partitionsToMakeFollower.put(tp,
- InitialFetchState(leaderEndPoint, partition.getLeaderEpoch, fetchOffset))
+ val state = info.partition.toLeaderAndIsrPartitionState(tp, isNew)
+ if (partition.makeFollower(state, offsetCheckpoints, Some(info.topicId))) {
+ Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
+ case None =>
+ stateChangeLogger.trace(
+ s"Unable to start fetching ${tp} with topic ID ${info.topicId} from leader " +
+ s"${leader} because it is not alive."
+ )
Review comment:
Created this issue to cover this case: https://issues.apache.org/jira/browse/KAFKA-13181
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11189: KAFKA-13161: Update replica partition state and replica fetcher state on follower update
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685631569
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2203,37 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
Review comment:
Done. Verified that the new tests fail without the changes to replica manager.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685452038
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2204,32 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
- // Create the local replica even if the leader is unavailable. This is required
- // to ensure that we include the partition's high watermark in the checkpoint
- // file (see KAFKA-1647)
- partition.createLogIfNotExists(isNew, false, offsetCheckpoints, Some(info.topicId))
-
if (shuttingDown) {
stateChangeLogger.trace(s"Unable to start fetching ${tp} with topic " +
s"ID ${info.topicId} because the replica manager is shutting down.")
} else {
val listenerName = config.interBrokerListenerName.value()
val leader = info.partition.leader
- Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
- case None => stateChangeLogger.trace(s"Unable to start fetching ${tp} " +
- s"with topic ID ${info.topicId} from leader ${leader} because it is not " +
- "alive.")
- case Some(node) =>
- val leaderEndPoint = new BrokerEndPoint(node.id(), node.host(), node.port())
- val log = partition.localLogOrException
- val fetchOffset = initialFetchOffset(log)
- partitionsToMakeFollower.put(tp,
- InitialFetchState(leaderEndPoint, partition.getLeaderEpoch, fetchOffset))
+ val state = info.partition.toLeaderAndIsrPartitionState(tp, isNew)
+ if (partition.makeFollower(state, offsetCheckpoints, Some(info.topicId))) {
+ Option(newImage.cluster().broker(leader)).flatMap(_.node(listenerName).asScala) match {
Review comment:
Intuitively, we should update `Partition` regardless whether the leader is available or not, but I was thinking it might be easier to do that separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #11189: KAFKA-13161: Always update the partition state
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685611958
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2203,37 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
Review comment:
Hmm, as I'm looking over the code, this seems wrong. We'd want to trigger delayed operations after the call to `makeFollower` since the change from leader to follower is what would allow the operation to complete.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11189: KAFKA-13161: Update replica partition state and replica fetcher state on follower update
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685624308
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2203,37 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
Review comment:
Yes. You are correct. Let me write a test for it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #11189: KAFKA-13161: Update replica partition state and replica fetcher state on follower update
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #11189:
URL: https://github.com/apache/kafka/pull/11189#discussion_r685624308
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -2204,27 +2203,37 @@ class ReplicaManager(val config: KafkaConfig,
completeDelayedFetchOrProduceRequests(tp)
Review comment:
Yes. You are correct. Let me write tests for it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #11189: KAFKA-13161: Update replica partition state and replica fetcher state on follower update
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #11189:
URL: https://github.com/apache/kafka/pull/11189
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org