CASSANDRA-21474: NoSpamLogger uses unbounded cache that could lead to memory exhaustion

src/java/org/apache/cassandra/config/CassandraRelevantProperties.java

@@ -406,6 +406,7 @@ public enum CassandraRelevantProperties
     MEMTABLE_OVERHEAD_SIZE("cassandra.memtable.row_overhead_size", "-1"),
     MEMTABLE_SHARD_COUNT("cassandra.memtable.shard.count"),
     MEMTABLE_TRIE_SIZE_LIMIT("cassandra.trie_size_limit_mb"),
+
     METRICS_REPORTER_CONFIG_FILE("cassandra.metricsReporterConfigFile"),
     /** Defines the maximum number of unique timed out queries that will be reported in the logs. Use a negative number to remove any limit. */
     MONITORING_MAX_OPERATIONS("cassandra.monitoring_max_operations", "50"),
@@ -428,6 +429,12 @@ public enum CassandraRelevantProperties
      * This is an optimization used in unit tests becuase we never restart a node there. The only node is stopoped
      * when the JVM terminates. Therefore, we can use such optimization and not wait unnecessarily. */
     NON_GRACEFUL_SHUTDOWN("cassandra.test.messagingService.nonGracefulShutdown"),
+    /**
+     * Maximum number of log statements cached per NoSpamLogger instance.
+     * This prevents unbounded memory growth when log messages contain dynamic content.
+     * Defaults to MAX_VALUE as a default behavior since we rely on the cache time-based expiration.
+     */
+    NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER("cassandra.nospam_logger.max_statements_per_logger", String.valueOf(Long.MAX_VALUE)),
     /** for specific tests */
     /** This property indicates whether disable_mbean_registration is true */
     ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION("org.apache.cassandra.disable_mbean_registration"),

src/java/org/apache/cassandra/utils/NoSpamLogger.java

@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.utils;
 
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
@@ -27,6 +28,12 @@
 import org.slf4j.Logger;
 
 import static org.apache.cassandra.utils.Clock.Global;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Expiry;
+import com.github.benmanes.caffeine.cache.Ticker;
+
+import static org.apache.cassandra.config.CassandraRelevantProperties.NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER;
 
 /**
  * Logging that limits each log statement to firing based on time since the statement last fired.
@@ -36,8 +43,11 @@
  * result in the original time being used. No warning is provided if there is a mismatch.
  *
  * If the statement is cached and used to log directly then only a volatile read will be required in the common case.
- * If the Logger is cached then there is a single concurrent hash map lookup + the volatile read.
- * If neither the logger nor the statement is cached then it is two concurrent hash map lookups + the volatile read.
+ * If the Logger is cached then there is a single Caffeine cache lookup + the volatile read.
+ * If neither the logger nor the statement is cached then it is a NonBlockingHashMap lookup + a Caffeine cache lookup + the volatile read.
+ *
+ * The implementation uses Caffeine cache with time-based expiration to automatically evict log statements
+ * after their minimum interval has passed, preventing unbounded memory growth from dynamic log messages.
  *
  */
 public class NoSpamLogger
@@ -64,6 +74,9 @@ public static void unsafeSetClock(Clock clock)
         CLOCK = clock;
     }
 
+    @VisibleForTesting
+    static Ticker TICKER = Ticker.systemTicker();
+
     public class NoSpamLogStatement extends AtomicLong
     {
         private static final long serialVersionUID = 1L;
@@ -157,6 +170,11 @@ public boolean error(Object... objects)
         {
             return NoSpamLogStatement.this.error(CLOCK.nanoTime(), objects);
         }
+
+        public long expiry()
+        {
+            return minIntervalNanos;
+        }
     }
 
     private static final NonBlockingHashMap<Logger, NoSpamLogger> wrappedLoggers = new NonBlockingHashMap<>();
@@ -167,6 +185,28 @@ static void clearWrappedLoggersForTest()
         wrappedLoggers.clear();
     }
 
+    /**
+     * Forces eviction of entries from the {@link NoSpamLogStatement} cache for this logger instance.
+     * This is useful for testing to ensure cache size limits are enforced immediately.
+     */
+    @VisibleForTesting
+    void cleanUpStatementsForTest()
+    {
+        lastMessage.cleanUp();
+    }
+
+    /**
+     * Returns the current size of the lastMessage cache for this logger instance.
+     * This is useful for testing cache eviction behavior.
+     *
+     * @return the number of log statements currently cached for this logger
+     */
+    @VisibleForTesting
+    long getStatementsCount()
+    {
+        return lastMessage.estimatedSize();
+    }
+
     public static NoSpamLogger getLogger(Logger logger, long minInterval, TimeUnit unit)
     {
         NoSpamLogger wrapped = wrappedLoggers.get(logger);
@@ -222,7 +262,41 @@ public static NoSpamLogStatement getStatement(Logger logger, String message, lon
 
     private final Logger wrapped;
     private final long minIntervalNanos;
-    private final NonBlockingHashMap<String, NoSpamLogStatement> lastMessage = new NonBlockingHashMap<>();
+
+    /**
+     * Cache of NoSpamLogStatement instances per NoSpamLogger instance.
+     * Bounded by size and time to prevent memory exhaustion from dynamic log messages.
+     * Uses Caffeine with W-TinyLFU eviction policy.
+     * Uses custom per-entry expiry based on each statement's minIntervalNanos.
+     */
+    private final Cache<String, NoSpamLogStatement> lastMessage = Caffeine.newBuilder()
+                                                                          .maximumSize(NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER.getLong())
+                                                                          .expireAfter(new Expiry<String, NoSpamLogStatement>()
+                                                                          {
+                                                                              @Override
+                                                                              public long expireAfterCreate(String key, NoSpamLogStatement value, long currentTime)
+                                                                              {
+                                                                                  return value.expiry();
+                                                                              }
+
+                                                                              @Override
+                                                                              public long expireAfterUpdate(String key, NoSpamLogStatement value,
+                                                                                                            long currentTime, long currentDuration)
+                                                                              {
+                                                                                  return value.expiry();
+                                                                              }
+
+                                                                              @Override
+                                                                              public long expireAfterRead(String key, NoSpamLogStatement value,
+                                                                                                          long currentTime, long currentDuration)
+                                                                              {
+                                                                                  return currentDuration;
+                                                                              }
+                                                                          })
+                                                                          .ticker(TICKER)
+                                                                          .executor(ForkJoinPool.commonPool())
+                                                                          .recordStats()
+                                                                          .build();
 
     private NoSpamLogger(Logger wrapped, long minInterval, TimeUnit timeUnit)
     {
@@ -302,14 +376,6 @@ public NoSpamLogStatement getStatement(String s, long minIntervalNanos)
 
     public NoSpamLogStatement getStatement(String key, String s, long minIntervalNanos)
     {
-        NoSpamLogStatement statement = lastMessage.get(key);
-        if (statement == null)
-        {
-            statement = new NoSpamLogStatement(s, minIntervalNanos);
-            NoSpamLogStatement temp = lastMessage.putIfAbsent(key, statement);
-            if (temp != null)
-                statement = temp;
-        }
-        return statement;
+        return lastMessage.get(key, k -> new NoSpamLogStatement(s, minIntervalNanos));
     }
 }