diff --git a/CHANGES.txt b/CHANGES.txt index 6b098881f8d2..10c27f93acf4 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java index 6b2913320143..b409fdcf09c7 100644 --- a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java +++ b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java @@ -114,8 +114,8 @@ public class ReplicaFilteringProtection> 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 @@ -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 @@ -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) diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java index fd8110cba72a..8ff974a0dd60 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java @@ -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; @@ -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)); } } @@ -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)); @@ -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)); @@ -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)); @@ -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 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)