Provide ability to configure IAuditLogger

Patch by Stefan Miklosovic; reviewed by Vinay Chella and marcuse for CASSANDRA-15748

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0-alpha5
+ * Provide ability to configure IAuditLogger (CASSANDRA-15748)
  * Fix nodetool enablefullquerylog blocking param parsing (CASSANDRA-15819)
  * Add isTransient to SSTableMetadataView (CASSANDRA-15806)
  * Fix tools/bin/fqltool for all shells (CASSANDRA-15820)

conf/cassandra.yaml

@@ -1284,7 +1284,8 @@ back_pressure_strategy:
 # on audit_logging for full details about the various configuration options.
 audit_logging_options:
     enabled: false
-    logger: BinAuditLogger
+    logger:
+      - class_name: BinAuditLogger
     # audit_logs_dir:
     # included_keyspaces:
     # excluded_keyspaces: system, system_schema, system_virtual_schema

doc/source/new/auditlogging.rst

@@ -120,7 +120,8 @@ The audit logger is set with the ``logger`` option.
 
 ::
 
- logger: BinAuditLogger
+ logger:
+ - class_name: BinAuditLogger
 
 Two types of audit loggers are supported: ``FileAuditLogger`` and ``BinAuditLogger``.
 ``BinAuditLogger`` is the default setting.  The ``BinAuditLogger`` is an efficient way to log events to file in a binary format.
@@ -129,6 +130,25 @@ Two types of audit loggers are supported: ``FileAuditLogger`` and ``BinAuditLogg
 
 The ``NoOpAuditLogger`` is a No-Op implementation of the audit logger to be used as a default audit logger when audit logging is disabled.
 
+It is possible to configure your custom logger implementation by injecting a map of property keys and their respective values. Default `IAuditLogger`
+implementations shipped with Cassandra do not react on these properties but your custom logger might. They would be present as
+a parameter of logger constructor (as `Map<String, String>`). In ``cassandra.yaml`` file, you may configure it like this:
+
+::
+
+ logger:
+ - class_name: MyCustomAuditLogger
+   parameters:
+   - key1: value1
+     key2: value2
+
+When it comes to configuring these parameters, you can use respective ``enableAuditLog`` method in ``StorageServiceMBean``.
+There are two methods of same name with different signatures. The first one does not accept a map where your parameters would be. This method
+is used primarily e.g. from JConsole or similar tooling. JConsole can not accept a map to be sent over JMX so in order to be able to enable it
+from there, even without any parameters, use this method. ``BinAuditLogger`` does not need any parameters to run with so invoking this method is fine.
+The second one does accept a map with your custom parameters so you can pass them programmatically. ``enableauditlog`` command of ``nodetool`` uses
+the first ``enableAuditLog`` method mentioned. Hence, currently, there is not a way how to pass parameters to your custom audit logger from ``nodetool``.
+
 Setting the Audit Logs Directory
 ********************************
 The audit logs directory is set with the ``audit_logs_dir`` option. A new directory is not created automatically and an existing directory must be set. Audit Logs directory can be configured using ``cassandra.logdir.audit`` system property or default is set to ``cassandra.logdir + /audit/``. A user created directory may be set. As an example, create a directory for the audit logs and set its permissions.
@@ -344,7 +364,8 @@ To demonstrate audit logging enable and configure audit logs with following sett
 
  audit_logging_options:
     enabled: true
-    logger: BinAuditLogger
+    logger:
+    - class_name: BinAuditLogger
     audit_logs_dir: "/cassandra/audit/logs/hourly"
     # included_keyspaces:
     # excluded_keyspaces: system, system_schema, system_virtual_schema

src/java/org/apache/cassandra/audit/AuditLogManager.java

@@ -20,7 +20,9 @@ package org.apache.cassandra.audit;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 
 import javax.annotation.Nullable;
@@ -31,6 +33,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.auth.AuthEvents;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.QueryEvents;
 import org.apache.cassandra.cql3.QueryOptions;
@@ -60,18 +63,20 @@ public class AuditLogManager implements QueryEvents.Listener, AuthEvents.Listene
 
     private AuditLogManager()
     {
-        if (DatabaseDescriptor.getAuditLoggingOptions().enabled)
+        final AuditLogOptions auditLogOptions = DatabaseDescriptor.getAuditLoggingOptions();
+
+        if (auditLogOptions.enabled)
         {
             logger.info("Audit logging is enabled.");
-            auditLogger = getAuditLogger(DatabaseDescriptor.getAuditLoggingOptions().logger);
+            auditLogger = getAuditLogger(auditLogOptions.logger);
         }
         else
         {
             logger.debug("Audit logging is disabled.");
-            auditLogger = new NoOpAuditLogger();
+            auditLogger = new NoOpAuditLogger(Collections.emptyMap());
         }
 
-        filter = AuditLogFilter.create(DatabaseDescriptor.getAuditLoggingOptions());
+        filter = AuditLogFilter.create(auditLogOptions);
     }
 
     public void initialize()
@@ -80,14 +85,14 @@ public class AuditLogManager implements QueryEvents.Listener, AuthEvents.Listene
             registerAsListener();
     }
 
-    private IAuditLogger getAuditLogger(String loggerClassName) throws ConfigurationException
+    private IAuditLogger getAuditLogger(ParameterizedClass logger) throws ConfigurationException
     {
-        if (loggerClassName != null)
+        if (logger.class_name != null)
         {
-            return FBUtilities.newAuditLogger(loggerClassName);
+            return FBUtilities.newAuditLogger(logger.class_name, logger.parameters == null ? Collections.emptyMap() : logger.parameters);
         }
 
-        return FBUtilities.newAuditLogger(BinAuditLogger.class.getName());
+        return FBUtilities.newAuditLogger(BinAuditLogger.class.getName(), Collections.emptyMap());
     }
 
     @VisibleForTesting
@@ -142,7 +147,7 @@ public class AuditLogManager implements QueryEvents.Listener, AuthEvents.Listene
     {
         unregisterAsListener();
         IAuditLogger oldLogger = auditLogger;
-        auditLogger = new NoOpAuditLogger();
+        auditLogger = new NoOpAuditLogger(Collections.emptyMap());
         oldLogger.stop();
     }
 
@@ -159,7 +164,7 @@ public class AuditLogManager implements QueryEvents.Listener, AuthEvents.Listene
         // next, check to see if we're changing the logging implementation; if not, keep the same instance and bail.
         // note: auditLogger should never be null
         IAuditLogger oldLogger = auditLogger;
-        if (oldLogger.getClass().getSimpleName().equals(auditLogOptions.logger))
+        if (oldLogger.getClass().getSimpleName().equals(auditLogOptions.logger.class_name))
             return;
 
         auditLogger = getAuditLogger(auditLogOptions.logger);

src/java/org/apache/cassandra/audit/AuditLogOptions.java

@@ -17,14 +17,18 @@
  */
 package org.apache.cassandra.audit;
 
+import java.util.Collections;
+import java.util.Map;
+
 import org.apache.commons.lang3.StringUtils;
 
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.utils.binlog.BinLogOptions;
 
 public class AuditLogOptions extends BinLogOptions
 {
     public volatile boolean enabled = false;
-    public String logger = BinAuditLogger.class.getSimpleName();
+    public ParameterizedClass logger = new ParameterizedClass(BinAuditLogger.class.getSimpleName(), Collections.emptyMap());
     public String included_keyspaces = StringUtils.EMPTY;
     // CASSANDRA-14498: By default, system, system_schema and system_virtual_schema are excluded, but these can be included via cassandra.yaml
     public String excluded_keyspaces = "system,system_schema,system_virtual_schema";

src/java/org/apache/cassandra/audit/BinAuditLogger.java

@@ -18,6 +18,7 @@
 package org.apache.cassandra.audit;
 
 import java.nio.file.Paths;
+import java.util.Map;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.primitives.Ints;
@@ -40,7 +41,7 @@ public class BinAuditLogger implements IAuditLogger
 
     private volatile BinLog binLog;
 
-    public BinAuditLogger()
+    public BinAuditLogger(Map<String, String> params)
     {
         AuditLogOptions auditLoggingOptions = DatabaseDescriptor.getAuditLoggingOptions();
 

src/java/org/apache/cassandra/audit/DiagnosticEventAuditLogger.java

@@ -18,10 +18,17 @@
 
 package org.apache.cassandra.audit;
 
+import java.util.Map;
+
 import org.apache.cassandra.diag.DiagnosticEventService;
 
 public class DiagnosticEventAuditLogger implements IAuditLogger
 {
+    public DiagnosticEventAuditLogger(Map<String, String> params)
+    {
+        
+    }
+
     public void log(AuditLogEntry logMessage)
     {
         AuditEvent.create(logMessage);

src/java/org/apache/cassandra/audit/FileAuditLogger.java

@@ -18,6 +18,8 @@
 
 package org.apache.cassandra.audit;
 
+import java.util.Map;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -30,7 +32,7 @@ public class FileAuditLogger implements IAuditLogger
 
     private volatile boolean enabled;
 
-    public FileAuditLogger()
+    public FileAuditLogger(Map<String, String> params)
     {
         enabled = true;
     }

src/java/org/apache/cassandra/audit/NoOpAuditLogger.java

@@ -17,11 +17,18 @@
  */
 package org.apache.cassandra.audit;
 
+import java.util.Map;
+
 /**
  * No-Op implementation of {@link IAuditLogger} to be used as a default audit logger when audit logging is disabled.
  */
 public class NoOpAuditLogger implements IAuditLogger
 {
+    public NoOpAuditLogger(Map<String, String> params)
+    {
+
+    }
+
     @Override
     public boolean isEnabled()
     {

src/java/org/apache/cassandra/db/virtual/SettingsTable.java

@@ -145,7 +145,7 @@ final class SettingsTable extends AbstractVirtualTable
 
         AuditLogOptions value = (AuditLogOptions) getValue(f);
         result.row(f.getName() + "_enabled").column(VALUE, Boolean.toString(value.enabled));
-        result.row(f.getName() + "_logger").column(VALUE, value.logger);
+        result.row(f.getName() + "_logger").column(VALUE, value.logger.class_name);
         result.row(f.getName() + "_audit_logs_dir").column(VALUE, value.audit_logs_dir);
         result.row(f.getName() + "_included_keyspaces").column(VALUE, value.included_keyspaces);
         result.row(f.getName() + "_excluded_keyspaces").column(VALUE, value.excluded_keyspaces);

src/java/org/apache/cassandra/service/StorageService.java

@@ -47,6 +47,7 @@ import com.google.common.base.Predicates;
 import com.google.common.collect.*;
 import com.google.common.util.concurrent.*;
 
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.dht.RangeStreamer.FetchReplica;
 import org.apache.cassandra.fql.FullQueryLogger;
 import org.apache.cassandra.fql.FullQueryLoggerOptions;
@@ -5444,14 +5445,20 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     public void enableAuditLog(String loggerName, String includedKeyspaces, String excludedKeyspaces, String includedCategories, String excludedCategories,
                                String includedUsers, String excludedUsers) throws ConfigurationException, IllegalStateException
     {
-        loggerName = loggerName != null ? loggerName : DatabaseDescriptor.getAuditLoggingOptions().logger;
+        enableAuditLog(loggerName, Collections.emptyMap(), includedKeyspaces, excludedKeyspaces, includedCategories, excludedCategories, includedUsers, excludedUsers);
+    }
+
+    public void enableAuditLog(String loggerName, Map<String, String> parameters, String includedKeyspaces, String excludedKeyspaces, String includedCategories, String excludedCategories,
+                               String includedUsers, String excludedUsers) throws ConfigurationException, IllegalStateException
+    {
+        loggerName = loggerName != null ? loggerName : DatabaseDescriptor.getAuditLoggingOptions().logger.class_name;
 
         Preconditions.checkNotNull(loggerName, "cassandra.yaml did not have logger in audit_logging_option and not set as parameter");
         Preconditions.checkState(FBUtilities.isAuditLoggerClassExists(loggerName), "Unable to find AuditLogger class: "+loggerName);
 
         AuditLogOptions auditLogOptions = new AuditLogOptions();
         auditLogOptions.enabled = true;
-        auditLogOptions.logger = loggerName;
+        auditLogOptions.logger = new ParameterizedClass(loggerName, parameters);
         auditLogOptions.included_keyspaces = includedKeyspaces != null ? includedKeyspaces : DatabaseDescriptor.getAuditLoggingOptions().included_keyspaces;
         auditLogOptions.excluded_keyspaces = excludedKeyspaces != null ? excludedKeyspaces : DatabaseDescriptor.getAuditLoggingOptions().excluded_keyspaces;
         auditLogOptions.included_categories = includedCategories != null ? includedCategories : DatabaseDescriptor.getAuditLoggingOptions().included_categories;
@@ -5463,7 +5470,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
 
         logger.info("AuditLog is enabled with logger: [{}], included_keyspaces: [{}], excluded_keyspaces: [{}], " +
                     "included_categories: [{}], excluded_categories: [{}], included_users: [{}], "
-                    + "excluded_users: [{}], archive_command: [{}]", loggerName, auditLogOptions.included_keyspaces, auditLogOptions.excluded_keyspaces,
+                    + "excluded_users: [{}], archive_command: [{}]", auditLogOptions.logger, auditLogOptions.included_keyspaces, auditLogOptions.excluded_keyspaces,
                     auditLogOptions.included_categories, auditLogOptions.excluded_categories, auditLogOptions.included_users, auditLogOptions.excluded_users,
                     auditLogOptions.archive_command);
 

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -740,6 +740,7 @@ public interface StorageServiceMBean extends NotificationEmitter
     /** Clears the history of clients that have connected in the past **/
     void clearConnectionHistory();
     public void disableAuditLog();
+    public void enableAuditLog(String loggerName, Map<String, String> parameters, String includedKeyspaces, String excludedKeyspaces, String includedCategories, String excludedCategories, String includedUsers, String excludedUsers) throws ConfigurationException;
     public void enableAuditLog(String loggerName, String includedKeyspaces, String excludedKeyspaces, String includedCategories, String excludedCategories, String includedUsers, String excludedUsers) throws ConfigurationException;
     public boolean isAuditLogEnabled();
     public String getCorruptedTombstoneStrategy();

src/java/org/apache/cassandra/tools/NodeProbe.java

@@ -1784,9 +1784,14 @@ public class NodeProbe implements AutoCloseable
         ssProxy.disableAuditLog();
     }
 
+    public void enableAuditLog(String loggerName, Map<String, String> parameters, String includedKeyspaces ,String excludedKeyspaces ,String includedCategories ,String excludedCategories ,String includedUsers ,String excludedUsers)
+    {
+        ssProxy.enableAuditLog(loggerName, parameters, includedKeyspaces, excludedKeyspaces, includedCategories, excludedCategories, includedUsers, excludedUsers);
+    }
+
     public void enableAuditLog(String loggerName, String includedKeyspaces ,String excludedKeyspaces ,String includedCategories ,String excludedCategories ,String includedUsers ,String excludedUsers)
     {
-        ssProxy.enableAuditLog(loggerName, includedKeyspaces, excludedKeyspaces, includedCategories, excludedCategories, includedUsers, excludedUsers);
+        this.enableAuditLog(loggerName, Collections.emptyMap(), includedKeyspaces, excludedKeyspaces, includedCategories, excludedCategories, includedUsers, excludedUsers);
     }
 
     public void enableOldProtocolVersions()

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

@@ -66,8 +66,6 @@ import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.io.util.DataOutputBufferFixed;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.net.AsyncOneResponse;
-
 
 
 public class FBUtilities
@@ -634,11 +632,20 @@ public class FBUtilities
         return FBUtilities.construct(className, "network authorizer");
     }
     
-    public static IAuditLogger newAuditLogger(String className) throws ConfigurationException
+    public static IAuditLogger newAuditLogger(String className, Map<String, String> parameters) throws ConfigurationException
     {
         if (!className.contains("."))
             className = "org.apache.cassandra.audit." + className;
-        return FBUtilities.construct(className, "Audit logger");
+
+        try
+        {
+            Class<?> auditLoggerClass = Class.forName(className);
+            return (IAuditLogger) auditLoggerClass.getConstructor(Map.class).newInstance(parameters);
+        }
+        catch (Exception ex)
+        {
+            throw new ConfigurationException("Unable to create instance of IAuditLogger.", ex);
+        }
     }
 
     public static boolean isAuditLoggerClassExists(String className)

test/unit/org/apache/cassandra/audit/AuditLoggerAuthTest.java

@@ -36,6 +36,7 @@ import com.datastax.driver.core.exceptions.UnauthorizedException;
 import org.apache.cassandra.OrderedJUnit4ClassRunner;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.OverrideConfigurationLoader;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.service.EmbeddedCassandraService;
@@ -70,7 +71,7 @@ public class AuditLoggerAuthTest
             config.role_manager = "CassandraRoleManager";
             config.authorizer = "CassandraAuthorizer";
             config.audit_logging_options.enabled = true;
-            config.audit_logging_options.logger = "InMemoryAuditLogger";
+            config.audit_logging_options.logger = new ParameterizedClass("InMemoryAuditLogger", null);
         });
         CQLTester.prepareServer();
 

test/unit/org/apache/cassandra/audit/AuditLoggerTest.java

@@ -21,6 +21,7 @@ import org.junit.After;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Map;
 
 import org.junit.Assert;
 import org.junit.Before;
@@ -36,6 +37,7 @@ import com.datastax.driver.core.exceptions.SyntaxError;
 import net.openhft.chronicle.queue.RollCycles;
 import org.apache.cassandra.auth.AuthEvents;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.cql3.QueryEvents;
 import org.apache.cassandra.exceptions.ConfigurationException;
@@ -61,7 +63,7 @@ public class AuditLoggerTest extends CQLTester
     {
         AuditLogOptions options = new AuditLogOptions();
         options.enabled = true;
-        options.logger = "InMemoryAuditLogger";
+        options.logger = new ParameterizedClass("InMemoryAuditLogger", null);
         DatabaseDescriptor.setAuditLoggingOptions(options);
         requireNetwork();
     }
@@ -89,7 +91,7 @@ public class AuditLoggerTest extends CQLTester
         String includedUsers = options.included_users;
         String excludedUsers = options.excluded_users;
 
-        StorageService.instance.enableAuditLog(loggerName, includedKeyspaces, excludedKeyspaces, includedCategories, excludedCategories, includedUsers, excludedUsers);
+        StorageService.instance.enableAuditLog(loggerName, null, includedKeyspaces, excludedKeyspaces, includedCategories, excludedCategories, includedUsers, excludedUsers);
     }
 
     private void disableAuditLogOptions()
@@ -676,7 +678,7 @@ public class AuditLoggerTest extends CQLTester
         disableAuditLogOptions();
         AuditLogOptions options = new AuditLogOptions();
         DatabaseDescriptor.setAuditLoggingOptions(options);
-        StorageService.instance.enableAuditLog(null, options.included_keyspaces, options.excluded_keyspaces, options.included_categories, options.excluded_categories, options.included_users, options.excluded_users);
+        StorageService.instance.enableAuditLog(null, null, options.included_keyspaces, options.excluded_keyspaces, options.included_categories, options.excluded_categories, options.included_users, options.excluded_users);
         try
         {
             assertEquals(1, QueryEvents.instance.listenerCount());
@@ -704,7 +706,7 @@ public class AuditLoggerTest extends CQLTester
         {
             assertEquals(1, QueryEvents.instance.listenerCount());
             assertEquals(0, AuthEvents.instance.listenerCount());
-            StorageService.instance.enableAuditLog(null, options.included_keyspaces, options.excluded_keyspaces, options.included_categories, options.excluded_categories, options.included_users, options.excluded_users);
+            StorageService.instance.enableAuditLog(null, null, options.included_keyspaces, options.excluded_keyspaces, options.included_categories, options.excluded_categories, options.included_users, options.excluded_users);
             fail("Conflicting directories - should throw exception");
         }
         catch (ConfigurationException e)

test/unit/org/apache/cassandra/audit/BinAuditLoggerTest.java

@@ -30,6 +30,7 @@ import net.openhft.chronicle.queue.ChronicleQueueBuilder;
 import net.openhft.chronicle.queue.ExcerptTailer;
 import net.openhft.chronicle.queue.RollCycles;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.utils.binlog.BinLogTest;
 
@@ -50,7 +51,7 @@ public class BinAuditLoggerTest extends CQLTester
 
         AuditLogOptions options = new AuditLogOptions();
         options.enabled = true;
-        options.logger = "BinAuditLogger";
+        options.logger = new ParameterizedClass("BinAuditLogger", null);
         options.roll_cycle = "TEST_SECONDLY";
         options.audit_logs_dir = tempDir.toString();
         DatabaseDescriptor.setAuditLoggingOptions(options);

test/unit/org/apache/cassandra/audit/InMemoryAuditLogger.java

@@ -17,8 +17,8 @@
  */
 package org.apache.cassandra.audit;
 
-import java.nio.file.Path;
 import java.util.LinkedList;
+import java.util.Map;
 import java.util.Queue;
 
 public class InMemoryAuditLogger implements IAuditLogger
@@ -26,6 +26,11 @@ public class InMemoryAuditLogger implements IAuditLogger
     final Queue<AuditLogEntry> inMemQueue = new LinkedList<>();
     private boolean enabled = true;
 
+    public InMemoryAuditLogger(Map<String, String> params)
+    {
+
+    }
+
     @Override
     public boolean isEnabled()
     {

test/unit/org/apache/cassandra/db/virtual/SettingsTableTest.java

@@ -34,6 +34,7 @@ import com.datastax.driver.core.ResultSet;
 import com.datastax.driver.core.Row;
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions.InternodeEncryption;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.CQLTester;
 
 public class SettingsTableTest extends CQLTester
@@ -186,7 +187,7 @@ public class SettingsTableTest extends CQLTester
         check(pre + "enabled", "true");
 
         check(pre + "logger", "BinAuditLogger");
-        config.audit_logging_options.logger = "logger";
+        config.audit_logging_options.logger = new ParameterizedClass("logger", null);
         check(pre + "logger", "logger");
 
         config.audit_logging_options.audit_logs_dir = "dir";

test/unit/org/apache/cassandra/service/StorageServiceServerTest.java

@@ -666,11 +666,11 @@ public class StorageServiceServerTest
     @Test
     public void testAuditLogEnableLoggerNotFound() throws Exception
     {
-        StorageService.instance.enableAuditLog(null, null, null, null, null, null, null);
+        StorageService.instance.enableAuditLog(null, null, null, null, null, null, null, null);
         assertTrue(AuditLogManager.instance.isEnabled());
         try
         {
-            StorageService.instance.enableAuditLog("foobar", null, null, null, null, null, null);
+            StorageService.instance.enableAuditLog("foobar", null, null, null, null, null, null, null);
             Assert.fail();
         }
         catch (IllegalStateException ex)
@@ -682,19 +682,19 @@ public class StorageServiceServerTest
     @Test
     public void testAuditLogEnableLoggerTransitions() throws Exception
     {
-        StorageService.instance.enableAuditLog(null, null, null, null, null, null, null);
+        StorageService.instance.enableAuditLog(null, null, null, null, null, null, null, null);
         assertTrue(AuditLogManager.instance.isEnabled());
 
         try
         {
-            StorageService.instance.enableAuditLog("foobar", null, null, null, null, null, null);
+            StorageService.instance.enableAuditLog("foobar", null, null, null, null, null, null, null);
         }
         catch (ConfigurationException | IllegalStateException e)
         {
             e.printStackTrace();
         }
 
-        StorageService.instance.enableAuditLog(null, null, null, null, null, null, null);
+        StorageService.instance.enableAuditLog(null, null, null, null, null, null, null, null);
         assertTrue(AuditLogManager.instance.isEnabled());
         StorageService.instance.disableAuditLog();
     }

test/unit/org/apache/cassandra/transport/CQLUserAuditTest.java

@@ -46,6 +46,7 @@ import org.apache.cassandra.audit.AuditLogEntryType;
 import org.apache.cassandra.audit.AuditLogManager;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.OverrideConfigurationLoader;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.diag.DiagnosticEventService;
 import org.apache.cassandra.locator.InetAddressAndPort;
@@ -69,7 +70,7 @@ public class CQLUserAuditTest
             config.role_manager = "CassandraRoleManager";
             config.diagnostic_events_enabled = true;
             config.audit_logging_options.enabled = true;
-            config.audit_logging_options.logger = "DiagnosticEventAuditLogger";
+            config.audit_logging_options.logger = new ParameterizedClass("DiagnosticEventAuditLogger", null);
         });
         CQLTester.prepareServer();