Skip to content

CASSANDRA-21623 Report the number of rows replica filtering protection cached when it warns #5162

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
AhmedReda85 wants to merge 1 commit into
base: trunk
Choose a base branch
from
+50 −22
Open

CASSANDRA-21623 Report the number of rows replica filtering protection cached when it warns #5162

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
7.0
* Report the number of rows replica filtering protection cached when it warns (CASSANDRA-21623)
* Allow CQLSSTableWriter to specify SSTable id generator to use (CASSANDRA-21012)
* Reject LIKE patterns with a wildcard (%) anywhere other than the start or end (CASSANDRA-21068)
* Support pluggable default role initialization (CASSANDRA-21546)
Expand Down
44 changes: 26 additions & 18 deletions src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,8 +114,8 @@ public class ReplicaFilteringProtection<E extends Endpoints<E>>
private final int cachedRowsWarnThreshold;
private final int cachedRowsFailThreshold;

/** Tracks whether or not we've already hit the warning threshold while evaluating a partition. */
private boolean hitWarningThreshold = false;
/** Tracks whether or not we've already hit the failure threshold while evaluating a partition. */
private boolean hitFailureThreshold = false;

private int currentRowsCached = 0; // tracks the current number of cached rows
private int maxRowsCached = 0; // tracks the high watermark for the number of cached rows
Expand Down Expand Up @@ -290,7 +290,23 @@ private class QueryMergeListener implements UnfilteredPartitionIterators.MergeLi
public void close()
{
// If we hit the failure threshold before consuming a single partition, record the current rows cached.
tableMetrics.rfpRowsCachedPerQuery.update(Math.max(currentRowsCached, maxRowsCached));
maxRowsCached = Math.max(currentRowsCached, maxRowsCached);
tableMetrics.rfpRowsCachedPerQuery.update(maxRowsCached);

// Check the cached rows warning threshold at the end of the query, so we can report the maximum number
// of cached rows we have had during the query.
if (!hitFailureThreshold && maxRowsCached > cachedRowsWarnThreshold)
{
String message =
String.format("Replica filtering protection has cached up to %d rows during query %s, " +
"which is over the warning threshold of %d rows defined by " +
"'cached_replica_rows_warn_threshold' in cassandra.yaml.",
maxRowsCached, command.toCQLString(), cachedRowsWarnThreshold);

ClientWarn.instance.warn(message);
oneMinuteLogger.warn(message);
Tracing.trace(message);
}
}

@Override
Expand Down Expand Up @@ -326,26 +342,18 @@ private void incrementCachedRows()

if (currentRowsCached == cachedRowsFailThreshold + 1)
{
String message = String.format("Replica filtering protection has cached over %d rows during query %s. " +
"(See 'cached_replica_rows_fail_threshold' in cassandra.yaml.)",
cachedRowsFailThreshold, command.toCQLString());
hitFailureThreshold = true;

String message =
String.format("Replica filtering protection has cached %d rows during query %s, " +
"which is over the failure threshold of %d rows defined by " +
"'cached_replica_rows_fail_threshold' in cassandra.yaml.",
currentRowsCached, command.toCQLString(), cachedRowsFailThreshold);

logger.error(message);
Tracing.trace(message);
throw new OverloadedException(message);
}
else if (currentRowsCached == cachedRowsWarnThreshold + 1 && !hitWarningThreshold)
{
hitWarningThreshold = true;

String message = String.format("Replica filtering protection has cached over %d rows during query %s. " +
"(See 'cached_replica_rows_warn_threshold' in cassandra.yaml.)",
cachedRowsWarnThreshold, command.toCQLString());

ClientWarn.instance.warn(message);
oneMinuteLogger.warn(message);
Tracing.trace(message);
}
}

private void releaseCachedRows(int count)
Expand Down
27 changes: 23 additions & 4 deletions test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.exceptions.OverloadedException;
import org.apache.cassandra.service.StorageService;
import org.assertj.core.api.Assertions;

import static org.apache.cassandra.config.ReplicaFilteringProtectionOptions.DEFAULT_FAIL_THRESHOLD;
import static org.apache.cassandra.config.ReplicaFilteringProtectionOptions.DEFAULT_WARN_THRESHOLD;
Expand Down Expand Up @@ -112,6 +113,12 @@ public void testMissedUpdatesAroundCachingFailThreshold()
catch (RuntimeException e)
{
assertEquals(e.getClass().getName(), OverloadedException.class.getName());
Assertions.assertThat(e)
.hasMessageStartingWith(String.format("Replica filtering protection has cached %d rows during query",
REPLICAS * ROWS_PER_PARTITION))
.hasMessageContaining(String.format("which is over the failure threshold of %d rows defined by " +
"'cached_replica_rows_fail_threshold' in cassandra.yaml.",
REPLICAS * ROWS_PER_PARTITION - 1));
}
}

Expand Down Expand Up @@ -147,7 +154,7 @@ private void testMissedUpdates(String tableName, int warnThreshold, int failThre
// of that row for all replicas.
SimpleQueryResult oldResult = cluster.coordinator(1).executeWithResult(query, ALL, "old", PARTITIONS * ROWS_PER_PARTITION);
assertRows(oldResult.toObjectArrays());
verifyWarningState(shouldWarn, oldResult);
verifyWarningState(shouldWarn, PARTITIONS * REPLICAS, warnThreshold, oldResult);

// We should have made 3 row "completion" requests.
assertEquals(PARTITIONS, protectionQueryCount(cluster.get(1), tableName));
Expand All @@ -170,7 +177,10 @@ private void testMissedUpdates(String tableName, int warnThreshold, int failThre
row(0, 0, "new"), row(0, 1, "new"), row(0, 2, "new"),
row(2, 0, "new"), row(2, 1, "new"), row(2, 2, "new"));

verifyWarningState(warnThreshold < REPLICAS * ROWS_PER_PARTITION, newResult);
verifyWarningState(warnThreshold < REPLICAS * ROWS_PER_PARTITION,
REPLICAS * ROWS_PER_PARTITION,
warnThreshold,
newResult);

// We still sould only have made 3 row "completion" requests, with no replica divergence in the last query.
assertEquals(PARTITIONS, protectionQueryCount(cluster.get(1), tableName));
Expand All @@ -193,7 +203,7 @@ private void testMissedUpdates(String tableName, int warnThreshold, int failThre
row(0, 0, "future"), row(0, 1, "future"), row(0, 2, "future"),
row(2, 0, "future"), row(2, 1, "future"), row(2, 2, "future"));

verifyWarningState(shouldWarn, futureResult);
verifyWarningState(shouldWarn, PARTITIONS * REPLICAS, warnThreshold, futureResult);

// We sould have made 3 more row "completion" requests.
assertEquals(PARTITIONS * 2, protectionQueryCount(cluster.get(1), tableName));
Expand All @@ -212,11 +222,20 @@ private void updateAllRowsOn(int node, String table, String value)
cluster.get(node).executeInternal("UPDATE " + table + " SET v = ? WHERE k = ? and c = ?", value, i, j);
}

private void verifyWarningState(boolean shouldWarn, SimpleQueryResult futureResult)
private void verifyWarningState(boolean shouldWarn, int expectedRows, int warnThreshold, SimpleQueryResult futureResult)
{
List<String> futureWarnings = futureResult.warnings();
assertEquals(shouldWarn, futureWarnings.stream().anyMatch(w -> w.contains("cached_replica_rows_warn_threshold")));
assertEquals(shouldWarn ? 1 : 0, futureWarnings.size());
if (shouldWarn)
{
Assertions.assertThat(futureWarnings.get(0))
.startsWith(String.format("Replica filtering protection has cached up to %d rows during query",
expectedRows))
.contains(String.format("which is over the warning threshold of %d rows defined by " +
"'cached_replica_rows_warn_threshold' in cassandra.yaml.",
warnThreshold));
}
}

private long protectionQueryCount(IInvokableInstance instance, String tableName)
Expand Down