CASSANDRA-20959: Add support for counter tables/queries #4541
Conversation
aparna0522
force-pushed
the
anaik-counters-support
branch
3 times, most recently
from
0ff447a to
0164723
Compare
src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
Outdated
Show resolved
Hide resolved
aparna0522
force-pushed
the
anaik-counters-support
branch
from
0164723 to
fada5f9
Compare
iamaleksey
left a comment
iamaleksey
left a comment
There was a problem hiding this comment.
Not fully done with the review
src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
Outdated
Show resolved
Hide resolved
src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
Outdated
Show resolved
Hide resolved
src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
Outdated
Show resolved
Hide resolved
src/java/org/apache/cassandra/replication/TrackedWriteRequest.java
Outdated
Show resolved
Hide resolved
| * @param handler the response handler | ||
| * @param coordinatorAckInfo optional coordinator info for forwarded writes (null for local coordinator) | ||
| */ | ||
| public static void sendToReplicasOnly(Mutation mutation, ReplicaPlan.ForWrite plan, TrackedWriteResponseHandler handler, ForwardedWrite.CoordinatorAckInfo coordinatorAckInfo) |
There was a problem hiding this comment.
You should be able to refactor applyLocallyAndSendToReplicas() to use sendToReplicasOnly(), with a few modifications - instead of duplicating the very much same logic in both these methods.
aparna0522
force-pushed
the
anaik-counters-support
branch
from
ee07f5e to
66bba91
Compare
| writeMetrics.localRequests.mark(); | ||
|
|
||
| MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, token); | ||
| mutation = mutation.withMutationId(id); |
There was a problem hiding this comment.
Now that I'm thinking about this again, I think it would be fine to restore this line, so the logic is shared between regular- and counter mutations, and undo the change to CounterMutation (so it will use the id() that we've just set here).
|
|
||
| MutationId id = MutationTrackingService.instance.nextMutationId(keyspaceName, token); | ||
|
|
||
| if (logger.isTraceEnabled()) |
There was a problem hiding this comment.
You don't need logger.isTraceEnabled() check here. We use isXEnabled() log checks in two scenarios:
- The arguments to the method are expensive computations (or allocate), or
- There are more than two parameters (excluding the format string) to pass. This is because there exist overloads for format string + 0, 1, 2 arguments, but anything beyond goes through a vararg method (e.g.
void trace(String var1, Object... var2)which has to allocate an array / potentially box some primitives).
This should be somewhere in the style guide, but I can't actually find it myself :\
| Preconditions.checkState(foundSelf, "Coordinator must be a replica"); | ||
|
|
||
| // Notify handler that local write succeeded (mutation was already applied before calling this method) | ||
| if (notifyHandlerForLocal) |
There was a problem hiding this comment.
I believe it would be better to mimic, for counters, in LocalCounterMutationRunnable#run() what we do for regular mutations, and invoke handler.onResponse(null) there, instead of having it be done two different ways for regular- and counter mutations. I'd also rename the method to simply sendToReplicas() - 'only' is ambiguous here.
| TrackedWriteRequest.sendToReplicasOnly(result, plan, handler, coordinatorAckInfo); | ||
|
|
||
| // Send this leader's response back to coordinator | ||
| MessagingService.instance().send(message.emptyResponse(), respondToAddress); |
There was a problem hiding this comment.
I think this code is missing setting up a callback on the leader. If you look at MutationVerbHandler#respond() you'll notice that the replicas respond to the coordinator (so the client can be notified) and to the leader (so that the leader can mark the id as witnessed on that replica pro-actively). For regular mutations we seem to be using LeaderCallback class for this.
| * @param coordinatorAckInfo optional coordinator info for forwarded writes (null for local coordinator) | ||
| * @param notifyHandlerForLocal if true, notify handler for local replica (used when mutation already applied) | ||
| */ | ||
| private static void sendToReplicasInternal(Mutation mutation, ReplicaPlan.ForWrite plan, |
There was a problem hiding this comment.
I believe it should be possible to reuse this method for ForwardedWrite#applyLocallyAndForwardToReplicas() now as well?
| writeMetrics.remoteRequests.mark(); | ||
| return ForwardedWrite.forwardMutation(mutation, plan, rs, requestTime); | ||
|
|
||
| if (mutation instanceof CounterMutation) |
There was a problem hiding this comment.
I personally would move this branching to ForwardedWrite, to a forward method that accepts an IMutation.
3 participants
Add support for counter tables/queries in Mutation Tracking
patch by Aparna Naik; for CASSANDRA-20959