CEP-45: Topology Change Support for Mutation Tracking#4910
Conversation
patch by Aleksey Yeschenko; reviewed by Blake Eggleston for CASSANDRA-20386
bdeggleston
left a comment
bdeggleston
left a comment
There was a problem hiding this comment.
I really like this design, very clean, practical, and minimal. There are a handful of implementation details that could use some refinement though. Most of this stuff has similar comments inline as well, sorry for the duplication.
Timeouts in the sealing coordinator.
We should rework all of the messaging by the SealingCoordiantor to use longer timeouts, as well as retries with backoff. Overly aggressive timeouts caused a lot of pain with the initial deployment of paxos repair for topology changes. Topology changes aren't as time critical as client requests, and failing a topology change causes a lot of operational headaches, especially since they are often done under duress. This can be a followup ticket, but it should be pretty simple to add
Finding a way to handle new logs without disrupting the write path.
We've discussed this in person and I've left a comment below. Adding a blocking network request in the write executor when we find a new log is not ideal. I think this could be a followup ticket
Inflight write bookkeeping.
The inflight write bookkeeping informing the is-drained check is fragile. I think we need a method of doing this that doesn't risk losing data if we double release or permananently block sealing if we don't release. There are a few places where I'm not totally sure we always properly release, and I think it's a hazard we should just eliminate. Off the top of my head, a rough idea would be to track outstanding issued mutation ids in the shard, release them in the apply path, and make an exploding mutation id variant that throws an exception if you try serializing it after some deadline has passed. If the outstanding id list isn't empty on drain, then wait past the deadline of the most recently issued id, I think it would fix both those bookeeping hazards. I think this should be addressed as part of this ticket
Cleanup.
Another thing we should consider doing is checking for shards in the SEALING state that should have been promoted to SEALED. We should assume that there are going to be state machine bugs that slip through, and having a periodic nanny task that cleans up hung state like that (things that can safely be cleaned up) will make bugs that would have been P0 fire drills into P1/P2 tickets. We should also have something like this for migration/satellite failover. This could be a followup ticket as well.
| /** | ||
| * Must be called exactly once per {@code nextId()} invocation. | ||
| */ | ||
| public void completeLocalWrite(MutationId id) |
There was a problem hiding this comment.
it's a little awkward having to remember remember to call this for each locally applied mutation, and any mistake is going to basically hang sealing which it waits for a decrement that never comes. Should we apply all local reads in a try with resources block or something?
| * Queries all peers for the shard metadata, finds or creates the matching shard locally, and returns it. | ||
| * Throws if resolution fails entirely. | ||
| */ | ||
| private ShardMetadata queryPeersForShardMetadata(CoordinatorLogId logId, String keyspace) |
There was a problem hiding this comment.
This is probably the right way to do this for log ids we don't know about, but I'm worried about this causing a flurry of requests and increase write latencies every time a log rolls over. I thin kwe should address this a few ways.
- Proactive communication from the allocator of the new log. Putting the range in every mutation id is probably overkill, but the node creating the log should immediately notify it's peers of a new log and it's ranges/epoch/keyspace, or at least attempt to. If it's possible to send the notification before we start issuing mutation ids that would be ideal, but at least sending them as soon as the log is created is probably good enough, so mutations don't have a period where there's an additional round trip on the apply process.
- The request/response process here should not be happening in the write stage. We should probably detect that we don't have the coordinator id and start the ShardMetadataRequest asynchronously, resubmitting the mutation when it's complete. We should also dedupe calls against a log id as well. One thing I'm not sure of is if that might cause problems with futures waiting on mutation apply?
There was a problem hiding this comment.
Also - this is kind of what we discussed earlier today, but I'm not saying this is the way to do it, just that we should refine this a bit, and this is one way to do it
|
|
||
| private static void drain(Set<ShardMetadata> shards, @Nullable NodeId withoutNode) | ||
| { | ||
| long deadlineNanos = nanoTime() + DatabaseDescriptor.getRpcTimeout(NANOSECONDS); |
There was a problem hiding this comment.
We should probably use the repair request timeout here, and maybe add some additional buffer with backoff and retry on top. Tight deadlines caused a lot of issues on paxos repair for topology changes and it's not time critical.
| private static void pollUntilWitnesses(ShardMetadata shard, Log2OffsetsMap.Mutable offsets, InetAddressAndPort endpoint) | ||
| { | ||
| // TODO (expected): use a longer timeout | ||
| long deadlineNanos = nanoTime() + DatabaseDescriptor.getRpcTimeout(NANOSECONDS); |
There was a problem hiding this comment.
Same timeout+retry+backoff comment here
| Request request = message.payload; | ||
| Log2OffsetsMap<?> witnessed = | ||
| MutationTrackingService.instance() | ||
| .collectLocallyWitnessedOffsets(request.keyspace, request.sinceEpoch, request.range); |
There was a problem hiding this comment.
I think we'd want to capture all offsets from all shards with a sinceEpoch <= request.sinceEpoch here? It may not be technically unneccesary, but it seems like a reasonable hedge against racing shard seal operations.
|
|
||
| public static AsyncPromise<Boolean> poll( | ||
| String keyspace, long sinceEpoch, Range<Token> range, Log2OffsetsMap.Mutable offsets, InetAddressAndPort endpoint) | ||
| { |
There was a problem hiding this comment.
| { | |
| { | |
| Preconditions.checkArgument(!endpoints.isEmpty()); |
| */ | ||
| public static AsyncPromise<Void> complete( | ||
| String keyspace, long sinceEpoch, Range<Token> range, List<InetAddressAndPort> endpoints) | ||
| { |
There was a problem hiding this comment.
| { | |
| { | |
| Preconditions.checkArgument(!endpoints.isEmpty()); |
| ClusterMetadataService.instance().ensureCMSPlacement(metadata); | ||
|
|
||
| if (MutationTrackingService.isEnabled()) | ||
| SealingCoordinator.sealShardsAtFinishReplace(metadata, finishReplace.delta(), startReplace.replaced()); |
There was a problem hiding this comment.
I think this should happen before we set the bootstrap state on the system keyspace? Otherwise a failure after the CMS commit could leave the shards in a SEALING state. I don't think there's downside to sealing before the commit. If it fails, the seal will be retried
| ClusterMetadataService.instance().ensureCMSPlacement(metadata); | ||
|
|
||
| if (MutationTrackingService.isEnabled()) | ||
| SealingCoordinator.sealShardsAtFinishJoin(metadata, finishJoin.delta()); |
There was a problem hiding this comment.
Same concern about dangling SEALING shards as in BootstrapAndReplace
| // during START_LEAVE, plus the merge-half folded into the departed node's range by the merge. | ||
| if (MutationTrackingService.isEnabled()) | ||
| if (streams.kind() == LeaveStreams.Kind.UNBOOTSTRAP || streams.kind() == LeaveStreams.Kind.REMOVENODE) | ||
| SealingCoordinator.sealShardsAtFinishLeave(postFinish, finishLeave.delta(), finishLeave.nodeId(), streams.kind()); |
There was a problem hiding this comment.
Same concern about dangling SEALING shards as in BootstrapAndReplace
stage, test, plus throw on rebuild for MT-enabled keyspaces
2 participants
patch by Aleksey Yeschenko; reviewed by Blake Eggleston for CASSANDRA-20386