You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "aweisberg (via GitHub)" <gi...@apache.org> on 2026/04/14 16:20:21 UTC
[PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
aweisberg opened a new pull request, #4740:
URL: https://github.com/apache/cassandra/pull/4740
https://issues.apache.org/jira/browse/CASSANDRA-21252
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "aweisberg (via GitHub)" <gi...@apache.org>.
aweisberg commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3126168670
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
Review Comment:
The list is bounded to the number of replicas we could potentially repair so it's not that large. It's also only called once as part of generating the errors when it times out.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3121562181
##########
src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java:
##########
@@ -119,10 +120,27 @@ void sendReadCommand(Replica to, ReadCallback<E, P> readCallback, boolean specul
abstract Meter getRepairMeter();
+ /**
+ * Mark the repair meter at most once per read command. Single-partition reads mark
+ * via {@code startRepair()} on digest mismatch. Range reads never call {@code startRepair()}
+ * — they go through {@code DataResolver.resolveWithReadRepair()} which calls
+ * {@code repairPartition()} directly for each inconsistent partition. Without this
+ * guard, range reads would either miss the metric entirely (if only marked in
+ * {@code startRepair()}) or over-count (if marked per partition in {@code repairPartition()}).
+ */
+ protected void markRepairMeter()
+ {
+ if (!repairMeterMarked)
Review Comment:
nit: We could use `AtomicBoolean` here, but I'm guessing the `volatile` is safe because we only need to guarantee visibility between multiple threads hitting this sequentially?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3121795951
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
Review Comment:
nit: Would iterating the full list of promises on ever call here ever become a problem?
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
Review Comment:
nit: Would iterating the full list of promises on every call here ever become a problem?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3139094599
##########
src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java:
##########
@@ -119,10 +120,27 @@ void sendReadCommand(Replica to, ReadCallback<E, P> readCallback, boolean specul
abstract Meter getRepairMeter();
+ /**
+ * Mark the repair meter at most once per read command. Single-partition reads mark
+ * via {@code startRepair()} on digest mismatch. Range reads never call {@code startRepair()}
+ * — they go through {@code DataResolver.resolveWithReadRepair()} which calls
+ * {@code repairPartition()} directly for each inconsistent partition. Without this
+ * guard, range reads would either miss the metric entirely (if only marked in
+ * {@code startRepair()}) or over-count (if marked per partition in {@code repairPartition()}).
+ */
+ protected void markRepairMeter()
+ {
+ if (!repairMeterMarked)
Review Comment:
Should we mark `AbstractReadRepair` with `@NotThreadSafe` in that case? Also noting for the record that we're also making `digestRepair` non-volatile, which is fine if we're indeed accessing from the same thread I guess.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "aweisberg (via GitHub)" <gi...@apache.org>.
aweisberg commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3126135686
##########
src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java:
##########
@@ -119,10 +120,27 @@ void sendReadCommand(Replica to, ReadCallback<E, P> readCallback, boolean specul
abstract Meter getRepairMeter();
+ /**
+ * Mark the repair meter at most once per read command. Single-partition reads mark
+ * via {@code startRepair()} on digest mismatch. Range reads never call {@code startRepair()}
+ * — they go through {@code DataResolver.resolveWithReadRepair()} which calls
+ * {@code repairPartition()} directly for each inconsistent partition. Without this
+ * guard, range reads would either miss the metric entirely (if only marked in
+ * {@code startRepair()}) or over-count (if marked per partition in {@code repairPartition()}).
+ */
+ protected void markRepairMeter()
+ {
+ if (!repairMeterMarked)
Review Comment:
It doesn't even need to be volatile since the same request thread is always the one that checks and sets this. I think it would be clearer to just remove 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "aweisberg (via GitHub)" <gi...@apache.org>.
aweisberg commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3126182564
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ if (!promise.isDone())
+ count++;
+ }
+ return count;
+ }
+ });
+ }
+
+ /**
+ * Start a tracked write attempt for a single read repair mutation.
+ * Wires a callback on the handler that resolves the promise on success, retries on retriable errors,
+ * or fails the promise on non-retriable errors.
+ *
+ * @param allowRetry if true, retriable errors (migration races) will refetch metadata and retry once
+ * through the correct path; if false, any error fails the promise directly
+ */
+ private void startTrackedWriteAttempt(DecoratedKey dk, Replica replica, Mutation mutation,
+ ConsistencyLevel cl, AsyncPromise<Void> promise, boolean allowRetry)
+ {
+ TrackedWriteRequest.perform(mutation, cl, requestTime, new WriteCallback()
+ {
+ @Override
+ public void run()
+ {
+ throw new UnsupportedOperationException("Should not be called");
Review Comment:
That would mean that lambda's no longer work for the callback without being changed to take in the parameter. Which wouldn't be that disruptive of a change so it's viable.
Originally it was runnable and the problem was that I could bind the promise to the closure that was in the callback. The solution here is to have the completion method pass in the promise without changing the signature for lambdas so other call sites don't have to change.
WDYT?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3121758687
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
Review Comment:
Do we actually need `blockFor` (and the overridden method below) if `awaitRepairs()` doesn't do anything with 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3121832105
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ if (!promise.isDone())
+ count++;
+ }
+ return count;
+ }
+ });
+ }
+
+ /**
+ * Start a tracked write attempt for a single read repair mutation.
+ * Wires a callback on the handler that resolves the promise on success, retries on retriable errors,
+ * or fails the promise on non-retriable errors.
+ *
+ * @param allowRetry if true, retriable errors (migration races) will refetch metadata and retry once
+ * through the correct path; if false, any error fails the promise directly
+ */
+ private void startTrackedWriteAttempt(DecoratedKey dk, Replica replica, Mutation mutation,
+ ConsistencyLevel cl, AsyncPromise<Void> promise, boolean allowRetry)
+ {
+ TrackedWriteRequest.perform(mutation, cl, requestTime, new WriteCallback()
+ {
+ @Override
+ public void run()
+ {
+ throw new UnsupportedOperationException("Should not be called");
+ }
+
+ @Override
+ public void onComplete(AbstractWriteResponseHandler<?> handler)
+ {
+ try
+ {
+ handler.get(); // non-blocking — condition already signaled when callback fires
+ promise.trySuccess(null);
+ }
+ catch (CoordinatorBehindException e)
+ {
+ if (allowRetry)
+ {
+ logger.debug("Tracked write read repair failed with retriable error, retrying: {}", e.getMessage());
+ retryRepairMutation(dk, replica, mutation, cl, promise);
+ }
+ else
+ {
+ logger.debug("Tracked write read repair retry failed again, not retrying further: {}", e.getMessage());
+ promise.tryFailure(e);
+ }
+ }
+ catch (Throwable t)
+ {
+ promise.tryFailure(t);
+ }
+ }
+ });
+ }
+
+ /**
+ * Retry a single read repair mutation after a migration race. Checks updated ClusterMetadata and
+ * sends the mutation through the correct path (tracked or untracked), resolving the promise
+ * when complete.
+ */
+ private void retryRepairMutation(DecoratedKey dk, Replica replica, Mutation mutation,
+ ConsistencyLevel cl, AsyncPromise<Void> promise)
+ {
+ ClusterMetadata cm = ClusterMetadata.current();
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ {
+ // Still tracked after refetch — retry as tracked write, but do not allow further retries
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, false);
Review Comment:
Will the retry generate a new mutation ID? Can this happen after the mutation ID from the first attempt has been applied only on some replicas? (If this is all possible, are there any implications for later read reconciliation, and if so, do we have to attempt to reuse the first mutation ID?)
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "aweisberg (via GitHub)" <gi...@apache.org>.
aweisberg commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3126182564
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ if (!promise.isDone())
+ count++;
+ }
+ return count;
+ }
+ });
+ }
+
+ /**
+ * Start a tracked write attempt for a single read repair mutation.
+ * Wires a callback on the handler that resolves the promise on success, retries on retriable errors,
+ * or fails the promise on non-retriable errors.
+ *
+ * @param allowRetry if true, retriable errors (migration races) will refetch metadata and retry once
+ * through the correct path; if false, any error fails the promise directly
+ */
+ private void startTrackedWriteAttempt(DecoratedKey dk, Replica replica, Mutation mutation,
+ ConsistencyLevel cl, AsyncPromise<Void> promise, boolean allowRetry)
+ {
+ TrackedWriteRequest.perform(mutation, cl, requestTime, new WriteCallback()
+ {
+ @Override
+ public void run()
+ {
+ throw new UnsupportedOperationException("Should not be called");
Review Comment:
Yeah I guess it would be since there is only one other caller. I'll do that. I thought there were more. I'll change 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3121785577
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ if (!promise.isDone())
+ count++;
+ }
+ return count;
+ }
+ });
+ }
+
+ /**
+ * Start a tracked write attempt for a single read repair mutation.
+ * Wires a callback on the handler that resolves the promise on success, retries on retriable errors,
+ * or fails the promise on non-retriable errors.
+ *
+ * @param allowRetry if true, retriable errors (migration races) will refetch metadata and retry once
+ * through the correct path; if false, any error fails the promise directly
+ */
+ private void startTrackedWriteAttempt(DecoratedKey dk, Replica replica, Mutation mutation,
+ ConsistencyLevel cl, AsyncPromise<Void> promise, boolean allowRetry)
+ {
+ TrackedWriteRequest.perform(mutation, cl, requestTime, new WriteCallback()
+ {
+ @Override
+ public void run()
+ {
+ throw new UnsupportedOperationException("Should not be called");
Review Comment:
nit: Would it be simpler to just get rid of `run()` and avoid this kind of thing (and throw an ignored `handler` argument in `wrapViewBatchResponseHandler()`, etc.)?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "aweisberg (via GitHub)" <gi...@apache.org>.
aweisberg commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3126205108
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
+
+ repairs.add(new PendingPartitionRepair()
+ {
+ @Override
+ public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException
+ {
+ long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining);
+ Throwable error = null;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ long remainingNanos = deadlineNanos - nanoTime();
+ if (remainingNanos <= 0)
+ return false;
+ try
+ {
+ promise.get(remainingNanos, NANOSECONDS);
+ }
+ catch (TimeoutException e)
+ {
+ return false;
+ }
+ catch (ExecutionException e)
+ {
+ error = Throwables.merge(error, e.getCause());
+ }
+ }
+ Throwables.maybeFail(error);
+ return true;
+ }
+
+ @Override
+ public ForWrite repairPlan()
+ {
+ return writePlan;
+ }
+
+ @Override
+ public int blockFor()
+ {
+ return blockFor;
+ }
+
+ @Override
+ public int waitingOn()
+ {
+ int count = 0;
+ for (AsyncPromise<Void> promise : promises)
+ {
+ if (!promise.isDone())
+ count++;
+ }
+ return count;
+ }
+ });
+ }
+
+ /**
+ * Start a tracked write attempt for a single read repair mutation.
+ * Wires a callback on the handler that resolves the promise on success, retries on retriable errors,
+ * or fails the promise on non-retriable errors.
+ *
+ * @param allowRetry if true, retriable errors (migration races) will refetch metadata and retry once
+ * through the correct path; if false, any error fails the promise directly
+ */
+ private void startTrackedWriteAttempt(DecoratedKey dk, Replica replica, Mutation mutation,
+ ConsistencyLevel cl, AsyncPromise<Void> promise, boolean allowRetry)
+ {
+ TrackedWriteRequest.perform(mutation, cl, requestTime, new WriteCallback()
+ {
+ @Override
+ public void run()
+ {
+ throw new UnsupportedOperationException("Should not be called");
+ }
+
+ @Override
+ public void onComplete(AbstractWriteResponseHandler<?> handler)
+ {
+ try
+ {
+ handler.get(); // non-blocking — condition already signaled when callback fires
+ promise.trySuccess(null);
+ }
+ catch (CoordinatorBehindException e)
+ {
+ if (allowRetry)
+ {
+ logger.debug("Tracked write read repair failed with retriable error, retrying: {}", e.getMessage());
+ retryRepairMutation(dk, replica, mutation, cl, promise);
+ }
+ else
+ {
+ logger.debug("Tracked write read repair retry failed again, not retrying further: {}", e.getMessage());
+ promise.tryFailure(e);
+ }
+ }
+ catch (Throwable t)
+ {
+ promise.tryFailure(t);
+ }
+ }
+ });
+ }
+
+ /**
+ * Retry a single read repair mutation after a migration race. Checks updated ClusterMetadata and
+ * sends the mutation through the correct path (tracked or untracked), resolving the promise
+ * when complete.
+ */
+ private void retryRepairMutation(DecoratedKey dk, Replica replica, Mutation mutation,
+ ConsistencyLevel cl, AsyncPromise<Void> promise)
+ {
+ ClusterMetadata cm = ClusterMetadata.current();
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ {
+ // Still tracked after refetch — retry as tracked write, but do not allow further retries
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, false);
Review Comment:
Hmm... if we started tracked and need to retry and it's tracked after I'm not sure we should retry. That seems like a it should be a checkState and fail instead. I'll change it to do that.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "aweisberg (via GitHub)" <gi...@apache.org>.
aweisberg commented on code in PR #4740:
URL: https://github.com/apache/cassandra/pull/4740#discussion_r3126159759
##########
src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java:
##########
@@ -209,12 +226,193 @@ public void repairPartition(DecoratedKey dk, Map<Replica, Mutation> mutations, R
// then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation
// which will ensure the repair mutation runs in the command store thread after preceding transactions are done
ClusterMetadata cm = ClusterMetadata.current();
- if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
+ if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken()))
+ repairViaTrackedWrite(dk, mutations, writePlan);
+ else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord))
repairViaAccordTransaction(dk, mutations, writePlan);
else
repairViaReadCoordinator(dk, mutations, writePlan, rrSource);
}
+ /*
+ * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform().
+ * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable.
+ * Each tracked write gets a proper MutationId and is recorded in the mutation journal.
+ *
+ * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced),
+ * we refetch ClusterMetadata and retry through the correct path (tracked or untracked).
+ */
+ private void repairViaTrackedWrite(DecoratedKey dk, Map<Replica, Mutation> mutations, ForWrite writePlan)
+ {
+ ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark();
+ ConsistencyLevel cl = writePlan.consistencyLevel();
+
+ List<AsyncPromise<Void>> promises = new ArrayList<>(mutations.size());
+
+ for (Map.Entry<Replica, Mutation> entry : mutations.entrySet())
+ {
+ Replica replica = entry.getKey();
+ Mutation mutation = entry.getValue();
+ AsyncPromise<Void> promise = new AsyncPromise<>();
+ promises.add(promise);
+ startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true);
+ }
+
+ // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted
+ // for contacts that don't have mutations to send
+ int adjustedBlockFor = writePlan.writeQuorum();
+ for (Replica participant : writePlan.contacts())
+ {
+ if (!mutations.containsKey(participant))
+ adjustedBlockFor--;
+ }
+ final int blockFor = adjustedBlockFor;
Review Comment:
It's used for diagnostic in `awaitWrites()`. It's not set for Accord because Accord's not really respecting your QUORUM request anyways in most cases. In this case we genuinely are going to use your CL and blockFor matters so it's included.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSANDRA-21252 CEP-45: BRR doesn't do tracked writes when needed during migration [cassandra]
Posted by "aweisberg (via GitHub)" <gi...@apache.org>.
aweisberg merged PR #4740:
URL: https://github.com/apache/cassandra/pull/4740
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org