Add HostedPartition.Fenced to ReplicaManager#494
Add HostedPartition.Fenced to ReplicaManager#494Ron Dagostino (rondagostino) wants to merge 20 commits into
Conversation
|
|
||
| // initially true when using a Raft-based metadata quorum, and may flip-flop thereafter; | ||
| // always false when using ZooKeeper | ||
| @volatile private var fenced: Boolean = zkClient.isEmpty |
There was a problem hiding this comment.
I am a little confused about the relationships between fenced, fencedPartitionStates, and allPartitions. I thought the advantage of having a Fenced state is that we would not need a separate collection because we can represent a fenced partition inside allPartitions. On the other hand, if fencing is really a global state as indicated by the flag here, then perhaps we need neither the Fenced state nor fencedPartitionStates. We could change getOnlinePartitionOrException to check for Online(partition) if !fenced.
Basically my concern is whether we can have clearly enforced invariants about the internal state here.
There was a problem hiding this comment.
I think we can eliminate fencedPartitionStates and keep the same information inside the HostedPartition.Fenced instance -- it does seem to be a waste to have both.
Regarding the fenced variable, I was thinking about how we would deal with becoming fenced during normal operations, well after we had started up and recovered our logs. In that situation, I did not intend this to represent a state such that every partition we lead or follow immediately transitions into the fenced state. For example, while we are fenced I think it is conceivable that we could continue to successfully fetch from other brokers for partitions that we believe we follow and successfully respond to fetch requests for partitions that we believe we lead. I intended this flag to indicate that any metadata changes related to partitions that we receive while in this state should not be relied upon as being the latest information and should be remembered for later (and potentially updated along the way) until we are no longer fenced and we know we have the right information. Basically, like at startup, the state indicates when we need to catch up on the metadata log before acting on any partition metadata we might receive while we are catching up.
So, to summarize what I was thinking:
- At startup it is true that every partition is fenced until after we initially catch up and recover logs, then we start leading/following
- Later on, if we become fenced, we could potentially still lead/follow for a while, but any new partition metadata that shows up isn't relied upon as being correct -- such partitions transition to the fenced state and the metadata is remembered for when we catch up again -- then we start leading/following as appropriate.
- There is a case where if we are fenced for too long it is possible that we might have to abandon our metadata entirely and fetch a new snapshot. This would happen if we are at metadata log offset X, but while fenced the offset X+1 became unavailable as an independently-fetchable offset and is only available as part of a snapshot. In this case we have to do a "soft restart" of sorts, and we have some reconciliation to do on our internal state. This reconciliation could be done in a number of ways, but one way would be to stop all leading/following, clear
allPartitions, set thefencedflag, and only clear thefencedflag after we have rwsd the latest snapshot and caught up.
| } | ||
| } | ||
|
|
||
| def localOnlineLogOrException(topicPartition: TopicPartition): Log = { |
There was a problem hiding this comment.
Could we please add a jmh benchmark for the frequently called ReplicaManager log lookup methods? Many of these are called very frequently in hot paths such as the fetch API code or replica fetcher, and we need to make sure we don't regress as the code is already overly heavy.
I had an experimental patch to cut down on the use of Either and it dropped the CPU time and garbage by half. Unfortunately I can't locate the experiment or I would point you to the benchmark. I'm fine if we don't change the approach here as long as we have the benchmark and we aren't regressing.
| } | ||
| stateChangeLogger.info(s"Fencing ${partitionsToBeFenced.size} partition(s)") | ||
| if (partitionsToBeFenced.nonEmpty) { | ||
| makeFenced(imageBuilder, partitionsToBeFenced, metadataOffset) |
There was a problem hiding this comment.
What happens to the onLeadershipChange callback if the partition is fenced? Because we may not be applying the change until later, I don't think we can send the callback through handleMetadataRecords. As an alternative, perhaps we need to provide a way to register a (say) LeadershipListener which gets notified about all leadership changes.
On a related note, we might need a notification to the GroupCoordinator when a partition becomes fenced so that it can stop attempting to write to the log. If we did a LeadershipListener, we could have three callbacks: onBecomeLeader, onBecomeFollower, and onBecomeFenced.
There was a problem hiding this comment.
I deferred the invocations of the onLeadershipChange callbacks so they get invoked after we apply the deferred changes.
Here's a counterargument to the suggestion that we notify when partition changes get deferred. We cannot rely on any received sate that we defer -- it may or may not be the correct state. We will eventually apply the correct state, but it may or may not be the initial state that we defer for a partition. I wonder if we want to act upon what could be intermediate state.
What we have may not be the ideal implementation, but I wonder if it at least correct n that the coordinators would not be able to incorrectly append to an offset/txn state partition.
There was a problem hiding this comment.
I've come around to assuming all bets are off once we are deferring partition state -- its easier to reason about, and the likeliest possibility by far is that we will xoon catch up and enter a consistent state. This means ideally we should notify right away when a partition first gets deferred state, and the receiver of the callback can decide what to do. Right now we only have onLeadershipChange with an indication if it's a new leader or a new follower. We might need something new like you suggest. However, since we are likely to be in a consistent state soon, maybe this is an optimization that can wait -- the "ideal" solution might not be to costly to defer (pun intended).
| } | ||
| } | ||
|
|
||
| // not sure if any of the 9 invocations of the above nonOfflinePartition() should be diverted here |
There was a problem hiding this comment.
Right. I don't think it's safe to expose the Partition object while it is in a fenced state. For example, GroupMetadataManager calls Partition.appendRecordsToLeader on the partition returned from nonOfflinePartition. But that would not be safe until the log has been initialized.
There was a problem hiding this comment.
would not be safe until the log has been initialized
What is unsafe about it? Looking at the code it throws NotLeaderOrFollowerException if the log doesn't exist, and if it does I assume it would be okay to append under the assumption that anything done incorrectly would not be committed. Is this incorrect?
There was a problem hiding this comment.
We now scope all "deferred" methods as private within ReplicaManager.
| } | ||
| } | ||
|
|
||
| def getNonOfflinePartitionOrException(topicPartition: TopicPartition): Partition = { |
There was a problem hiding this comment.
Similar to other comment. Can we refuse to expose Partition while it is fenced?
There was a problem hiding this comment.
Are we going to allow follower fetching when the broker is fenced? if so, then I wonder if it becomes necessary to expose the Partition even when we have deferred changes.
There was a problem hiding this comment.
If the broker is fenced then we won't be getting any new metadata, and nothing will defer. Everything related to deferred partitions is now scoped private within ReplicaManager.
| partitionsToMakeFollower | ||
| } | ||
|
|
||
| def makeDeferred(builder: MetadataImageBuilder, |
There was a problem hiding this comment.
Wondering if we need to complete purgatory operations on fencing as well.
|
In the current patch, the "deferred/fenced" state is not propagated to the An alternative we can consider is pushing the "deferred/fenced" state into |
|
|
||
| // Returns the Log for the partition if it is deferred or online and has one, otherwise raises an exception. | ||
| // Equivalent to localOnlineLogOrException() when using ZooKeeper. | ||
| def localDeferredOrOnlineLogOrException(topicPartition: TopicPartition): Log = { |
There was a problem hiding this comment.
I didn't see a response to my previous comment, so I will repeat it here in case it was missed. I think we need to be clear about the contract when the partition is in the "deferred" state. It's unclear to me why we need to expose either the Log or Partition object at all in this state. Keep in mind that in this patch when the state is "deferred," the leader/epoch information inside Partition is guaranteed to be out of date. It is only inside applyDeferredMetadataChanges that we apply the latest metadata to Partition. It may be safe to rely on the old state, but what is the point?
If ultimately we think this is the right implementation--that is if ultimately we just want to continue with the old state until the initialization/reloading has completed--then I'm not sure I see the benefit of mixing the deferred state into allPartitions. It would be simpler to collect the deferred entries in a separate map which would be applied at the end when the loading is complete. Then we could dump all of these "deferredOrOnline" APIs and potentially the duplication inside applyDeferredMetadataChanges.
There was a problem hiding this comment.
Metadata gets deferred when the broker is fenced and new metadata arrives. If new metadata is arriving then that probably means we are heartbeating successfully, and that likely means we'll be caught up soon.
I doubt we would end up in some intermediate state for very long. I mean, it's possible, but it's unlikely -- we would have to be fenced, then back in communication, then out of communication. The most likely event by far is that we have intermediate states for a short period of time.
So maybe the best thing to do is to assume the worst as soon as anything defers -- i.e. we don't expose the partition when it has deferred metadata.
| stateChangeLogger.info(s"Applied ${partitionsMadeLeader.size + partitionsMadeFollower.size} deferred partitions prior to the error: " + | ||
| s"${partitionsMadeLeader.size} leader(s) and ${partitionsMadeFollower.size} follower(s)") | ||
| // Re-throw the exception for it to be caught in BrokerMetadataListener | ||
| throw e |
There was a problem hiding this comment.
If we fail to apply changes, I guess we have to see that as a fatal error? The only possible way of recovering would be to replay the changes.
There was a problem hiding this comment.
If we fail to apply changes, I guess we have to see that as a fatal error? The only possible way of recovering would be to replay the changes.
Yeah, I think so. We may need to put effort into minimizing the blast radius of these failures.
I added this comment to the AK PR apache#10069
| val partitionsMadeLeader = makeLeaders(partitionsAlreadyExisting, leaderPartitionStates, | ||
| highWatermarkCheckpoints,-1, mostRecentMetadataOffsets) | ||
| val partitionsMadeFollower = makeFollowers(partitionsAlreadyExisting, | ||
| metadataCache.currentImage().brokers, followerPartitionStates, |
There was a problem hiding this comment.
Do we have any guarantee that the metadata cache is in a state that is consistent with the deferred changes?
There was a problem hiding this comment.
Do we have any guarantee that the metadata cache is in a state that is consistent with the deferred changes?
Good question. Metadata changes up through the point of applying the deferred partition metadata changes should be applied to the metadata cache at this point.
One thing we need to think about is the fact that we currently don't defer metadata cache changes at all. The metadata cache will contain partition states that are ahead of ReplicaManager during the time when ReplicaManager is deferring its changes. This means, for example, that the following will reflect deferred partition changes that have been applied to the metadata cache but that have not been applied to ReplicaManager. We may have to write test cases for each of these conditions so we can be clear on what the expected behavior should be.
- MetadataRequest
- FindCoordinatorRequest
- ElectLeadersRequest with topicPartitions = null
- DelayedCreatePartitions (in topic purgatory)
- DelayedElectLeader (in elect leader purgatory)
- Anything that calls ReplicaManager.fetchMessages() and DelayedFetch (in fetch purgatory), though these seem okay since they wait until they can get enough data?
- TransactionMarkerChannelManager.addTxnMarkersToBrokerQueue
- DescribeConfigsRequest
- OffsetCommitRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- ProduceRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- FetchRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- DeleteRecordsRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- AddPartitionsToTxnRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- TxnOffsetCommitRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- OffsetDeleteRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
I added this comment to the AK PR apache#10069
Ron Dagostino (rondagostino)
left a comment
Ron Dagostino (rondagostino)
left a comment
There was a problem hiding this comment.
This is superseded by the AK PR apache#10069.
| val partitionsMadeLeader = makeLeaders(partitionsAlreadyExisting, leaderPartitionStates, | ||
| highWatermarkCheckpoints,-1, mostRecentMetadataOffsets) | ||
| val partitionsMadeFollower = makeFollowers(partitionsAlreadyExisting, | ||
| metadataCache.currentImage().brokers, followerPartitionStates, |
There was a problem hiding this comment.
Do we have any guarantee that the metadata cache is in a state that is consistent with the deferred changes?
Good question. Metadata changes up through the point of applying the deferred partition metadata changes should be applied to the metadata cache at this point.
One thing we need to think about is the fact that we currently don't defer metadata cache changes at all. The metadata cache will contain partition states that are ahead of ReplicaManager during the time when ReplicaManager is deferring its changes. This means, for example, that the following will reflect deferred partition changes that have been applied to the metadata cache but that have not been applied to ReplicaManager. We may have to write test cases for each of these conditions so we can be clear on what the expected behavior should be.
- MetadataRequest
- FindCoordinatorRequest
- ElectLeadersRequest with topicPartitions = null
- DelayedCreatePartitions (in topic purgatory)
- DelayedElectLeader (in elect leader purgatory)
- Anything that calls ReplicaManager.fetchMessages() and DelayedFetch (in fetch purgatory), though these seem okay since they wait until they can get enough data?
- TransactionMarkerChannelManager.addTxnMarkersToBrokerQueue
- DescribeConfigsRequest
- OffsetCommitRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- ProduceRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- FetchRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- DeleteRecordsRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- AddPartitionsToTxnRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- TxnOffsetCommitRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
- OffsetDeleteRequest (whether or not to send UNKNOWN_TOPIC_OR_PARTITION)
I added this comment to the AK PR apache#10069
| stateChangeLogger.info(s"Applied ${partitionsMadeLeader.size + partitionsMadeFollower.size} deferred partitions prior to the error: " + | ||
| s"${partitionsMadeLeader.size} leader(s) and ${partitionsMadeFollower.size} follower(s)") | ||
| // Re-throw the exception for it to be caught in BrokerMetadataListener | ||
| throw e |
There was a problem hiding this comment.
If we fail to apply changes, I guess we have to see that as a fatal error? The only possible way of recovering would be to replay the changes.
Yeah, I think so. We may need to put effort into minimizing the blast radius of these failures.
I added this comment to the AK PR apache#10069
|
Closing in favor of the AK PR apache#10069 |
3 participants
As we catch up on the metadata log during startup, and before we have performed log recovery, we will assign partitions to a new "Fenced" state in ReplicaManager. This allows us to "remember" what we processed as we were catching up. Then later, after log recovery is complete, but before the broker becomes unfenced, we will iterate through all of the fenced partitions and become leader/follower as required.
Everything is implemented except the last parts -- getting notification that it is time to unfence the partitions and the implementation of the actual unfencing.
Committer Checklist (excluded from commit message)