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/01/29 15:12:58 UTC
[GitHub] [kafka] rondagostino opened a new pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
rondagostino opened a new pull request #10003:
URL: https://github.com/apache/kafka/pull/10003
Brokers receive metadata from the Raft metadata quorum very differently than they do from ZooKeeper today, and this has implications for ReplicaManager. In particular, when a broker reads the metadata log it may not arrive at the ultimate state for a partition until it reads multiple messages. In normal operation the multiple messages associated with a state change will all appear in a single batch, so they can and will be coalesced and applied together. There are circumstances where messages associated with partition state changes will appear across multiple batches and we will be forced to coalesce these multiple batches together. The circumstances when this occurs are as follows:
- When the broker restarts it must "catch up" on the metadata log, and it is likely that the broker will see multiple partition state changes for a single partition across different batches while it is catching up. For example, it will see the `TopicRecord` and the `PartitionRecords` for the topic creation, and then it will see any `IsrChangeRecords` that may have been recorded since the creation. The broker does not know the state of the topic partitions until it reads and coalesces all the messages.
- The broker will have to "catch up" on the metadata log if it becomes fenced and then regains its lease and resumes communication with the metadata quorum.
- A fenced broker may ultimately have to perform a "soft restart" if it was fenced for so long that the point at which it needs to resume fetching the metadata log has been subsumed into a metadata snapshot and is no longer independently fetchable. A soft restart will entail some kind of metadata reset based on the latest available snapshot plus a catchup phase to fetch after the snapshot end point.
The first case -- during startup -- occurs before clients are able to connect to the broker. Clients are able to connect to the broker in the second case. It is unclear if clients will be able to to connect to the broker during a soft restart (the third case).
We need a way to defer the application of topic partition metadata in all of the above cases, and while we are deferring the application of the metadata the broker will not service clients for the affected partitions.
As a side note, it is arguable if the broker should be able to service clients while catching up or not. The decision to not service clients has no impact in the startup case -- clients can't connect yet at that point anyway. In the third case it is not yet clear what we are going to do, but being unable to service clients while performing a soft reset seems reasonable. In the second case it is most likely true that we will catch up quickly; it would be unusual to reestablish communication with the metadata quorum such that we gain a new lease and begin to catch up only to lose our lease again.
So we need a way to defer the application of partition metadata and make those partitions unavailable while deferring state changes. This PR adds a new internal partition state to ReplicaManager to accomplish this. Currently the available partition states are simple `Online`, `Offline` (meaning a log dir failure) and `None` (meaning we don't know about it). We add a new `Deferred` state. We also rename a couple of methods that refer to "nonOffline" partitions to instead refer to "online" partitions.
**The new `Deferred` state never happens when using ZooKeeper for metadata storage.** Partitions can only enter the `Deferred` state when using a KIP-500 Raft metadata quorum and one of the above 3 cases occurs. The testing strategy is therefore to leverage existing tests to confirm that there is no functionality change in the ZooKeeper case. We will add the logic for deferring/applying/reacting to deferred partition state in separate PRs since that code will never be invoked in the ZooKeeper world.
### 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568653073
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
@hachikuji Have you considered having a sealed trait with `def partition: Partition` and then have the relevant case classes inherit from that? Then, when you want to get a partition, you match against that subtrait while knowing that the partition is always available in those cases. I suggested this offline to @rondagostino, but not sure if he didn't think it was useful or didn't get around to 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568862000
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -891,7 +944,7 @@ class ReplicaManager(val config: KafkaConfig,
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
val logStartOffset = getPartition(topicPartition) match {
case HostedPartition.Online(partition) => partition.logStartOffset
- case HostedPartition.None | HostedPartition.Offline => -1L
+ case _ => -1L
Review comment:
I think you want to call `onlinePartition` instead of `getPartition`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568971416
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -942,7 +942,7 @@ class ReplicaManager(val config: KafkaConfig,
requiredAcks: Short): Map[TopicPartition, LogAppendResult] = {
val traceEnabled = isTraceEnabled
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
- val logStartOffset = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
+ val logStartOffset: Long = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
Review comment:
Might be simpler to use -1L?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
ijuma commented on pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#issuecomment-772357616
A single unrelated flaky test failed for JDK 8:
https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-10003/10/
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r567065946
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -197,7 +204,8 @@ class ReplicaManager(val config: KafkaConfig,
val delayedDeleteRecordsPurgatory: DelayedOperationPurgatory[DelayedDeleteRecords],
val delayedElectLeaderPurgatory: DelayedOperationPurgatory[DelayedElectLeader],
threadNamePrefix: Option[String],
- val alterIsrManager: AlterIsrManager) extends Logging with KafkaMetricsGroup {
+ val alterIsrManager: AlterIsrManager,
+ usingRaftMetadataQuorum: Boolean) extends Logging with KafkaMetricsGroup {
Review comment:
Could we check if we're in Raft mode using KafkaConfig rather than adding a flag here?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568653073
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
@hachikuji Have you considered having a sealed trait with `def partition: Partition` and then have the relevant case classes inherit from that? Then, when you want to get a partition, you match against that subtrait while knowing that the partition is always available in those cases. I suggested this offline to @rondagostino, but not sure if he didn't think it was useful or didn't get around to it.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
After this change, we can probably keep `HostedPartition.None | HostedPartition.Offline` as it was before, right?
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -891,7 +944,7 @@ class ReplicaManager(val config: KafkaConfig,
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
val logStartOffset = getPartition(topicPartition) match {
case HostedPartition.Online(partition) => partition.logStartOffset
- case HostedPartition.None | HostedPartition.Offline => -1L
+ case _ => -1L
Review comment:
I think you want to call `onlinePartition` instead of `getPartition`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568215734
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
This is one case we probably _do_ want to act in the deferred state. One option I was considering is to expose `Partition` from `HostedPartition` as an option.
```scala
sealed trait HostedPartition {
def partition: Option[Partition]
}
```
Then we could refactor this as a `filter` or something.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568216372
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
+ stateChangeLogger.info(s"Metadata changes deferred=$deferringMetadataChanges")
+
+ private def confirmNotDeferringMetadataUpdatesWithZooKeeper(): Unit = {
+ if (deferringMetadataChanges) {
+ throw new IllegalStateException("Partition metadata changes should never be deferred when using ZooKeeper")
+ }
+ }
+
+ def beginMetadataChangeDeferral(): Unit = {
+ replicaStateChangeLock synchronized {
+ if (config.requiresZookeeper) {
+ throw new IllegalStateException("Partition metadata changes can never be deferred when using ZooKeeper")
+ }
+ deferringMetadataChanges = true
+ stateChangeLogger.info(s"Metadata changes are now being deferred")
+ }
+ }
+
+ def endMetadataChangeDeferral(): Unit = {
+ replicaStateChangeLock synchronized {
+ // TODO: implement for Raft-based metadata log messages
Review comment:
nit: we can leave the TODO out.
Also, it might be worth checking `config.requiresZookeeper` as we do in `beginMetadataChangeDeferral`. If nothing else, it serves as explicit documentation.
##########
File path: core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
##########
@@ -111,7 +111,7 @@ class ReplicaAlterLogDirsThread(name: String,
override def processPartitionData(topicPartition: TopicPartition,
fetchOffset: Long,
partitionData: PartitionData[Records]): Option[LogAppendInfo] = {
- val partition = replicaMgr.nonOfflinePartition(topicPartition).get
+ val partition = replicaMgr.onlinePartition(topicPartition).get
Review comment:
It seems like we should be using `getPartitionOrException` here. Similarly in `ReplicaFetcherThread`.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -1257,7 +1307,8 @@ class ReplicaManager(val config: KafkaConfig,
def maybeUpdateMetadataCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest) : Seq[TopicPartition] = {
replicaStateChangeLock synchronized {
- if(updateMetadataRequest.controllerEpoch < controllerEpoch) {
+ confirmNotDeferringMetadataUpdatesWithZooKeeper()
Review comment:
I think the check here and in `becomeLeaderOrFollower` is overkill. We already ensure that `beginMetadataChangeDeferral` cannot be called when using zk.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
This is one case we probably _do_ want to act in the deferred state. One option I was consider is whether we should give
```scala
sealed trait HostedPartition {
def partition: Option[Partition]
}
```
Then we could refactor this as a `filter` or something.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
+ stateChangeLogger.info(s"Metadata changes deferred=$deferringMetadataChanges")
+
+ private def confirmNotDeferringMetadataUpdatesWithZooKeeper(): Unit = {
+ if (deferringMetadataChanges) {
+ throw new IllegalStateException("Partition metadata changes should never be deferred when using ZooKeeper")
+ }
+ }
+
+ def beginMetadataChangeDeferral(): Unit = {
+ replicaStateChangeLock synchronized {
+ if (config.requiresZookeeper) {
Review comment:
nit: doesn't matter too much, but we may as well pull this outside the lock
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
+ stateChangeLogger.info(s"Metadata changes deferred=$deferringMetadataChanges")
Review comment:
I think we can remove this or lower to debug. It doesn't seem too useful.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -713,6 +760,9 @@ class ReplicaManager(val config: KafkaConfig,
case HostedPartition.Offline =>
throw new KafkaStorageException(s"Partition $topicPartition is offline")
+ case HostedPartition.Deferred(_) =>
+ throw new KafkaStorageException(s"Partition $topicPartition is deferred")
Review comment:
Hmm, this doesn't seem right. Not sure why a deferred change would cause a storage error. How about we throw `IllegalStateException` instead to make it clear that this is an unexpected state for now.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
Review comment:
Why does this need to volatile?
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
This is one case we probably _do_ want to act in the deferred state. One option I was consider is whether we should expose `Partition` from `HostedPartition` as an option.
```scala
sealed trait HostedPartition {
def partition: Option[Partition]
}
```
Then we could refactor this as a `filter` or something.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
This is one case we probably _do_ want to act in the deferred state. One option I was considering is to expose `Partition` from `HostedPartition` as an option.
```scala
sealed trait HostedPartition {
def partition: Option[Partition]
}
```
Then we could refactor this as a `filter` or something.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -891,7 +944,7 @@ class ReplicaManager(val config: KafkaConfig,
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
val logStartOffset = getPartition(topicPartition) match {
case HostedPartition.Online(partition) => partition.logStartOffset
- case HostedPartition.None | HostedPartition.Offline => -1L
+ case _ => -1L
Review comment:
Agreed. Maybe `onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1L)`?
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -942,7 +942,7 @@ class ReplicaManager(val config: KafkaConfig,
requiredAcks: Short): Map[TopicPartition, LogAppendResult] = {
val traceEnabled = isTraceEnabled
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
- val logStartOffset = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
+ val logStartOffset: Long = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
Review comment:
Might be simpler to use -1L?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568216372
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
+ stateChangeLogger.info(s"Metadata changes deferred=$deferringMetadataChanges")
+
+ private def confirmNotDeferringMetadataUpdatesWithZooKeeper(): Unit = {
+ if (deferringMetadataChanges) {
+ throw new IllegalStateException("Partition metadata changes should never be deferred when using ZooKeeper")
+ }
+ }
+
+ def beginMetadataChangeDeferral(): Unit = {
+ replicaStateChangeLock synchronized {
+ if (config.requiresZookeeper) {
+ throw new IllegalStateException("Partition metadata changes can never be deferred when using ZooKeeper")
+ }
+ deferringMetadataChanges = true
+ stateChangeLogger.info(s"Metadata changes are now being deferred")
+ }
+ }
+
+ def endMetadataChangeDeferral(): Unit = {
+ replicaStateChangeLock synchronized {
+ // TODO: implement for Raft-based metadata log messages
Review comment:
nit: we can leave the TODO out.
Also, it might be worth checking `config.requiresZookeeper` as we do in `beginMetadataChangeDeferral`. If nothing else, it serves as explicit documentation.
##########
File path: core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
##########
@@ -111,7 +111,7 @@ class ReplicaAlterLogDirsThread(name: String,
override def processPartitionData(topicPartition: TopicPartition,
fetchOffset: Long,
partitionData: PartitionData[Records]): Option[LogAppendInfo] = {
- val partition = replicaMgr.nonOfflinePartition(topicPartition).get
+ val partition = replicaMgr.onlinePartition(topicPartition).get
Review comment:
It seems like we should be using `getPartitionOrException` here. Similarly in `ReplicaFetcherThread`.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -1257,7 +1307,8 @@ class ReplicaManager(val config: KafkaConfig,
def maybeUpdateMetadataCache(correlationId: Int, updateMetadataRequest: UpdateMetadataRequest) : Seq[TopicPartition] = {
replicaStateChangeLock synchronized {
- if(updateMetadataRequest.controllerEpoch < controllerEpoch) {
+ confirmNotDeferringMetadataUpdatesWithZooKeeper()
Review comment:
I think the check here and in `becomeLeaderOrFollower` is overkill. We already ensure that `beginMetadataChangeDeferral` cannot be called when using zk.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
This is one case we probably _do_ want to act in the deferred state. One option I was consider is whether we should give
```scala
sealed trait HostedPartition {
def partition: Option[Partition]
}
```
Then we could refactor this as a `filter` or something.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
+ stateChangeLogger.info(s"Metadata changes deferred=$deferringMetadataChanges")
+
+ private def confirmNotDeferringMetadataUpdatesWithZooKeeper(): Unit = {
+ if (deferringMetadataChanges) {
+ throw new IllegalStateException("Partition metadata changes should never be deferred when using ZooKeeper")
+ }
+ }
+
+ def beginMetadataChangeDeferral(): Unit = {
+ replicaStateChangeLock synchronized {
+ if (config.requiresZookeeper) {
Review comment:
nit: doesn't matter too much, but we may as well pull this outside the lock
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
+ stateChangeLogger.info(s"Metadata changes deferred=$deferringMetadataChanges")
Review comment:
I think we can remove this or lower to debug. It doesn't seem too useful.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -713,6 +760,9 @@ class ReplicaManager(val config: KafkaConfig,
case HostedPartition.Offline =>
throw new KafkaStorageException(s"Partition $topicPartition is offline")
+ case HostedPartition.Deferred(_) =>
+ throw new KafkaStorageException(s"Partition $topicPartition is deferred")
Review comment:
Hmm, this doesn't seem right. Not sure why a deferred change would cause a storage error. How about we throw `IllegalStateException` instead to make it clear that this is an unexpected state for now.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
Review comment:
Why does this need to volatile?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r567064852
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +265,25 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferrable when using a Raft-based metadata quorum, and may flip-flop thereafter;
+ // changes are never deferrable when using ZooKeeper. When true, this indicates that we should transition online
Review comment:
Is the code that will make use of the Deferred case coming in a followup PR?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568730408
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
Review comment:
Good point. The same thread will be both managing this value and submitting metadata changes, so it can remember if it has already started deferring changes if it needs to do so; plus everything here related to deferral happens within the lock anyway. I removed `@volatile`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r566901097
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
Using the wildcard means you won't get any help if you add any new HostedPartition values. Worth thinking if that's what we want as a general pattern or if we prefer to enumerate and/or extract to other methods.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568756317
##########
File path: core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
##########
@@ -111,7 +111,7 @@ class ReplicaAlterLogDirsThread(name: String,
override def processPartitionData(topicPartition: TopicPartition,
fetchOffset: Long,
partitionData: PartitionData[Records]): Option[LogAppendInfo] = {
- val partition = replicaMgr.nonOfflinePartition(topicPartition).get
+ val partition = replicaMgr.onlinePartition(topicPartition).get
Review comment:
Yes, this appears to be an existing issue. I fixed it in 4 places.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r567070588
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +265,25 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferrable when using a Raft-based metadata quorum, and may flip-flop thereafter;
+ // changes are never deferrable when using ZooKeeper. When true, this indicates that we should transition online
+ // partitions to the deferred state if we see metadata with a different leader epoch.
+ @volatile private var changesDeferrable: Boolean = usingRaftMetadataQuorum
+ stateChangeLogger.info(s"Metadata changes deferrable=$changesDeferrable")
+
+ private def confirmChangesNotDeferrableWithZooKeeper(): Unit = {
+ if (changesDeferrable) {
+ throw new IllegalStateException("Partition metadata changes should never be deferrable when using ZooKeeper")
+ }
+ }
+
+ def deferrableMetadataChanges(): Unit = {
+ replicaStateChangeLock synchronized {
Review comment:
Is it the case that whenever we grab this lock we want to assert that we're in a legal state wrt ZK and `changesDeferrable`? If that's the case, we might want to make a note of it where we define `replicaStateChangeLock`.
Alternatively, maybe we could put this behind some helper function that grabs the lock and checks our state.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568716273
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
> a sealed trait with def partition: Partition
I like this solution better than making everything a case class with an `Option[Partition]`. Will push a commit with it shortly.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568894890
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -891,7 +944,7 @@ class ReplicaManager(val config: KafkaConfig,
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
val logStartOffset = getPartition(topicPartition) match {
case HostedPartition.Online(partition) => partition.logStartOffset
- case HostedPartition.None | HostedPartition.Offline => -1L
+ case _ => -1L
Review comment:
Agreed. Maybe `onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1L)`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568215734
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
This is one case we probably _do_ want to act in the deferred state. One option I was consider is whether we should expose `Partition` from `HostedPartition` as an option.
```scala
sealed trait HostedPartition {
def partition: Option[Partition]
}
```
Then we could refactor this as a `filter` or something.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568716273
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
> a sealed trait with def partition: Partition
I like this solution better than making everything a case class with an `Option[Partition]`. Will push a commit with it shortly.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +264,37 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferred when using a Raft-based metadata quorum, and they may flip-flop to not
+ // being deferred and being deferred again thereafter as the broker (re)acquires/loses its lease.
+ // Changes are never deferred when using ZooKeeper. When true, this indicates that we should transition
+ // online partitions to the deferred state if we see a metadata update for that partition.
+ @volatile private var deferringMetadataChanges: Boolean = !config.requiresZookeeper
Review comment:
Good point. The same thread will be both managing this value and submitting metadata changes, so it can remember if it has already started deferring changes if it needs to do so; plus everything here related to deferral happens within the lock anyway. I removed `@volatile`.
##########
File path: core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala
##########
@@ -111,7 +111,7 @@ class ReplicaAlterLogDirsThread(name: String,
override def processPartitionData(topicPartition: TopicPartition,
fetchOffset: Long,
partitionData: PartitionData[Records]): Option[LogAppendInfo] = {
- val partition = replicaMgr.nonOfflinePartition(topicPartition).get
+ val partition = replicaMgr.onlinePartition(topicPartition).get
Review comment:
Yes, this appears to be an existing issue. I fixed it in 4 places.
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -942,7 +942,7 @@ class ReplicaManager(val config: KafkaConfig,
requiredAcks: Short): Map[TopicPartition, LogAppendResult] = {
val traceEnabled = isTraceEnabled
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
- val logStartOffset = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
+ val logStartOffset: Long = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
Review comment:
Doh! I was confused about why it had failed. Will fix.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
ijuma commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568860838
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -316,7 +324,7 @@ class ReplicaManager(val config: KafkaConfig,
private def maybeRemoveTopicMetrics(topic: String): Unit = {
val topicHasOnlinePartition = allPartitions.values.exists {
case HostedPartition.Online(partition) => topic == partition.topic
- case HostedPartition.None | HostedPartition.Offline => false
+ case _ => false
Review comment:
After this change, we can probably keep `HostedPartition.None | HostedPartition.Offline` as it was before, right?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r567468635
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -257,6 +265,25 @@ class ReplicaManager(val config: KafkaConfig,
}
}
+ // Changes are initially deferrable when using a Raft-based metadata quorum, and may flip-flop thereafter;
+ // changes are never deferrable when using ZooKeeper. When true, this indicates that we should transition online
+ // partitions to the deferred state if we see metadata with a different leader epoch.
+ @volatile private var changesDeferrable: Boolean = usingRaftMetadataQuorum
+ stateChangeLogger.info(s"Metadata changes deferrable=$changesDeferrable")
+
+ private def confirmChangesNotDeferrableWithZooKeeper(): Unit = {
+ if (changesDeferrable) {
+ throw new IllegalStateException("Partition metadata changes should never be deferrable when using ZooKeeper")
+ }
+ }
+
+ def deferrableMetadataChanges(): Unit = {
+ replicaStateChangeLock synchronized {
Review comment:
It isn't entirely clear since there are places where this lock is acquired that we won't implement for the initial version of a Raft-based metadata quorum (e.g. handling log dir failure). I think it's a good point that we should check here, though, because we must never make changes deferrable if we are using ZooKeeper. So I added the check.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ijuma merged pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
ijuma merged pull request #10003:
URL: https://github.com/apache/kafka/pull/10003
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10003: MINOR: Add HostedPartition.Deferred state in ReplicaManager
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10003:
URL: https://github.com/apache/kafka/pull/10003#discussion_r568973909
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -942,7 +942,7 @@ class ReplicaManager(val config: KafkaConfig,
requiredAcks: Short): Map[TopicPartition, LogAppendResult] = {
val traceEnabled = isTraceEnabled
def processFailedRecord(topicPartition: TopicPartition, t: Throwable) = {
- val logStartOffset = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
+ val logStartOffset: Long = onlinePartition(topicPartition).map(_.logStartOffset).getOrElse(-1)
Review comment:
Doh! I was confused about why it had failed. Will fix.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org