Migrate logging API from SLF4J to Log4j 2.x (#459)

* Migrate logging API from SLF4J to Log4j 2.x

Standardize all modules on the Log4j 2.x API so Flume's own logging no
longer goes through the SLF4J facade. Applied the OpenRewrite
Slf4jToLog4j recipe, then normalized the result: drop loggers that were
never used, make every logger field private static final named "logger",
and use the no-arg LogManager.getLogger() since each one named its own
class.

Assisted-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Formatting

* Ban transitive logging implementations and bridges

Dependencies frequently ship a logging backend or an API bridge as a
compile dependency, which then leaks to our consumers. Add a
maven-enforcer rule that bans every known logging implementation/bridge
and whitelists only the real logging APIs.

To satisfy the rule, drop the now-unneeded log4j-1.2-api (no Flume code
path reaches the Hadoop classes that use the org.apache.log4j API) and
exclude the logging backends that Hadoop (reload4j) and ZooKeeper
(logback) pull in. Exclusions live in dependencyManagement so they apply
wherever the dependency is used, and only where the artifact is actually
pulled; the enforcer rule catches anything reintroduced later.

Assisted-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* Pin SLF4J to 2.x and use the SLF4J 2 binding

The ban-logging rule whitelisted the whole log4j2 group, which let the
SLF4J 1.x binding (log4j-slf4j-impl) through. Switch the test bindings to
log4j-slf4j2-impl and import slf4j-bom so conflict resolution cannot
downgrade transitive SLF4J to 1.x.

Drop the log4j-jul and log4j-jcl bridges the tightened rule then flagged,
and manage commons-logging to a current release so transitive deps stop
pulling the ancient 1.2.

Assisted-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ppkarwasz

flume-ng-auth/pom.xml

@@ -44,20 +44,14 @@
       <scope>test</scope>
     </dependency>
 
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-api</artifactId>
-    </dependency>
-
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
-      <artifactId>log4j-slf4j-impl</artifactId>
-      <scope>test</scope>
+      <artifactId>log4j-api</artifactId>
     </dependency>
 
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
-      <artifactId>log4j-1.2-api</artifactId>
+      <artifactId>log4j-slf4j2-impl</artifactId>
       <scope>test</scope>
     </dependency>
 
@@ -77,6 +71,7 @@
       <artifactId>hadoop-minikdc</artifactId>
       <scope>test</scope>
     </dependency>
+
     <dependency>
       <!-- add this to satisfy the dependency requirement of apacheds-jdbm1 in minikdc-->
       <groupId>org.apache.directory.jdbm</groupId>

flume-ng-auth/src/main/java/org/apache/flume/auth/KerberosAuthenticator.java

@@ -33,16 +33,16 @@
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 /**
  * A kerberos authenticator, which authenticates using the supplied principal
  * and keytab and executes with  authenticated privileges
  */
 class KerberosAuthenticator implements FlumeAuthenticator {
 
-    private static final Logger LOG = LoggerFactory.getLogger(KerberosAuthenticator.class);
+    private static final Logger logger = LogManager.getLogger();
 
     private volatile UserGroupInformation ugi;
     private volatile KerberosUser prevUser;
@@ -146,7 +146,7 @@ public synchronized void authenticate(String principal, String keytab) {
                 curUser = null;
             }
         } catch (IOException e) {
-            LOG.warn("User unexpectedly had no active login. Continuing with " + "authentication", e);
+            logger.warn("User unexpectedly had no active login. Continuing with " + "authentication", e);
         }
 
         /*
@@ -158,13 +158,13 @@ public synchronized void authenticate(String principal, String keytab) {
         try {
             if (ugi != null) {
                 if (curUser != null && curUser.getUserName().equals(ugi.getUserName())) {
-                    LOG.debug("Using existing principal login: {}", ugi);
+                    logger.debug("Using existing principal login: {}", ugi);
                 } else {
-                    LOG.info("Attempting kerberos Re-login as principal ({}) ", new Object[] {ugi.getUserName()});
+                    logger.info("Attempting kerberos Re-login as principal ({}) ", new Object[] {ugi.getUserName()});
                     ugi.reloginFromKeytab();
                 }
             } else {
-                LOG.info(
+                logger.info(
                         "Attempting kerberos login as principal ({}) from keytab " + "file ({})",
                         new Object[] {resolvedPrincipal, keytab});
                 UserGroupInformation.loginUserFromKeytab(resolvedPrincipal, keytab);
@@ -187,7 +187,7 @@ private void printUGI(UserGroupInformation ugi) {
         if (ugi != null) {
             // dump login information
             AuthenticationMethod authMethod = ugi.getAuthenticationMethod();
-            LOG.info("\n{} \nUser: {} \nAuth method: {} \nKeytab: {} \n", new Object[] {
+            logger.info("\n{} \nUser: {} \nAuth method: {} \nKeytab: {} \n", new Object[] {
                 authMethod.equals(AuthenticationMethod.PROXY) ? "Proxy as: " : "Logged as: ",
                 ugi.getUserName(),
                 authMethod,
@@ -213,7 +213,7 @@ public void run() {
                         try {
                             ugi.checkTGTAndReloginFromKeytab();
                         } catch (IOException e) {
-                            LOG.warn(
+                            logger.warn(
                                     "Error occured during checkTGTAndReloginFromKeytab() for user " + ugi.getUserName(),
                                     e);
                         }

flume-ng-channels/flume-file-channel/pom.xml

@@ -53,8 +53,8 @@
     </dependency>
 
     <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-api</artifactId>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-api</artifactId>
     </dependency>
 
     <dependency>
@@ -87,19 +87,7 @@
 
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
-      <artifactId>log4j-jcl</artifactId>
-      <scope>test</scope>
-    </dependency>
-
-    <dependency>
-      <groupId>org.apache.logging.log4j</groupId>
-      <artifactId>log4j-slf4j-impl</artifactId>
-      <scope>test</scope>
-    </dependency>
-
-    <dependency>
-      <groupId>org.apache.logging.log4j</groupId>
-      <artifactId>log4j-1.2-api</artifactId>
+      <artifactId>log4j-slf4j2-impl</artifactId>
       <scope>test</scope>
     </dependency>
 

flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/CheckpointRebuilder.java

@@ -32,8 +32,8 @@
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 public class CheckpointRebuilder {
 
@@ -45,7 +45,7 @@ public class CheckpointRebuilder {
     private final SetMultimap<Long, ComparableFlumeEventPointer> uncommittedTakes = HashMultimap.create();
     private final boolean fsyncPerTransaction;
 
-    private static Logger LOG = LoggerFactory.getLogger(CheckpointRebuilder.class);
+    private static final Logger logger = LogManager.getLogger();
 
     public CheckpointRebuilder(List<File> logFiles, FlumeEventQueue queue, boolean fsyncPerTransaction)
             throws IOException {
@@ -55,13 +55,13 @@ public CheckpointRebuilder(List<File> logFiles, FlumeEventQueue queue, boolean f
     }
 
     public boolean rebuild() throws IOException, Exception {
-        LOG.info("Attempting to fast replay the log files.");
+        logger.info("Attempting to fast replay the log files.");
         List<LogFile.SequentialReader> logReaders = Lists.newArrayList();
         for (File logFile : logFiles) {
             try {
                 logReaders.add(LogFileFactory.getSequentialReader(logFile, null, fsyncPerTransaction));
             } catch (EOFException e) {
-                LOG.warn("Ignoring " + logFile + " due to EOF", e);
+                logger.warn("Ignoring " + logFile + " due to EOF", e);
             }
         }
         long transactionIDSeed = 0;
@@ -120,7 +120,7 @@ public boolean rebuild() throws IOException, Exception {
                 }
             }
         } catch (Exception e) {
-            LOG.warn("Error while generating checkpoint using fast generation logic", e);
+            logger.warn("Error while generating checkpoint using fast generation logic", e);
             return false;
         } finally {
             TransactionIDOracle.setSeed(transactionIDSeed);
@@ -135,7 +135,7 @@ public boolean rebuild() throws IOException, Exception {
             queue.addTail(put.pointer);
             count++;
         }
-        LOG.info("Replayed {} events using fast replay logic.", count);
+        logger.info("Replayed {} events using fast replay logic.", count);
         return true;
     }
 
@@ -155,7 +155,7 @@ private void writeCheckpoint() throws IOException {
                 }
             }
         } catch (Exception e) {
-            LOG.warn("Error while generating checkpoint using fast generation logic", e);
+            logger.warn("Error while generating checkpoint using fast generation logic", e);
         } finally {
             for (LogFile.MetaDataWriter metaDataWriter : metaDataWriters) {
                 metaDataWriter.close();
@@ -228,7 +228,7 @@ public static void main(String[] args) throws Exception {
         int capacity = Integer.parseInt(cli.getOptionValue("t"));
         File checkpointFile = new File(checkpointDir, "checkpoint");
         if (checkpointFile.exists()) {
-            LOG.error("Cannot execute fast replay", new IllegalStateException("Checkpoint exists" + checkpointFile));
+            logger.error("Cannot execute fast replay", new IllegalStateException("Checkpoint exists" + checkpointFile));
         } else {
             EventQueueBackingStore backingStore = EventQueueBackingStoreFactory.get(
                     checkpointFile, capacity, "channel", new FileChannelCounter("Main"));
@@ -241,7 +241,7 @@ public static void main(String[] args) throws Exception {
             if (rebuilder.rebuild()) {
                 rebuilder.writeCheckpoint();
             } else {
-                LOG.error("Could not rebuild the checkpoint due to errors.");
+                logger.error("Could not rebuild the checkpoint due to errors.");
             }
         }
     }

flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFactory.java

@@ -21,11 +21,11 @@
 import java.io.IOException;
 import java.io.RandomAccessFile;
 import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 class EventQueueBackingStoreFactory {
-    private static final Logger LOG = LoggerFactory.getLogger(EventQueueBackingStoreFactory.class);
+    private static final Logger logger = LogManager.getLogger();
 
     private EventQueueBackingStoreFactory() {}
 
@@ -60,7 +60,7 @@ static EventQueueBackingStore get(
                 // delete everything in the checkpoint directory and force
                 // a full replay.
                 if (!checkpointExists || checkpointFile.length() == 0) {
-                    LOG.warn("MetaData file for checkpoint "
+                    logger.warn("MetaData file for checkpoint "
                             + " exists but checkpoint does not. Checkpoint = " + checkpointFile
                             + ", metaDataFile = " + metaDataFile);
                     throw new BadCheckpointException("The last checkpoint was not completed correctly, "
@@ -90,15 +90,15 @@ static EventQueueBackingStore get(
                 }
                 return new EventQueueBackingStoreFileV2(checkpointFile, capacity, name, counter);
             }
-            LOG.error("Found version " + Integer.toHexString(version) + " in " + checkpointFile);
+            logger.error("Found version " + Integer.toHexString(version) + " in " + checkpointFile);
             throw new BadCheckpointException(
                     "Checkpoint file exists with " + Serialization.VERSION_3 + " but no metadata file found.");
         } finally {
             if (checkpointFileHandle != null) {
                 try {
                     checkpointFileHandle.close();
                 } catch (IOException e) {
-                    LOG.warn("Unable to close " + checkpointFile, e);
+                    logger.warn("Unable to close " + checkpointFile, e);
                 }
             }
         }
@@ -113,7 +113,7 @@ private static EventQueueBackingStore upgrade(
             boolean compressBackup,
             FileChannelCounter counter)
             throws Exception {
-        LOG.info("Attempting upgrade of " + checkpointFile + " for " + name);
+        logger.info("Attempting upgrade of " + checkpointFile + " for " + name);
         EventQueueBackingStoreFileV2 backingStoreV2 =
                 new EventQueueBackingStoreFileV2(checkpointFile, capacity, name, counter);
         String backupName = checkpointFile.getName() + "-backup-" + System.currentTimeMillis();

flume-ng-channels/flume-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFile.java

@@ -38,11 +38,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.flume.channel.file.instrumentation.FileChannelCounter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 abstract class EventQueueBackingStoreFile extends EventQueueBackingStore {
-    private static final Logger LOG = LoggerFactory.getLogger(EventQueueBackingStoreFile.class);
+    private static final Logger logger = LogManager.getLogger();
     private static final int MAX_ALLOC_BUFFER_SIZE = 2 * 1024 * 1024; // 2MB
     protected static final int HEADER_SIZE = 1029;
     protected static final int INDEX_VERSION = 0;
@@ -94,7 +94,7 @@ protected EventQueueBackingStoreFile(
             checkpointFileHandle.seek(INDEX_VERSION * Serialization.SIZE_OF_LONG);
             checkpointFileHandle.writeLong(getVersion());
             checkpointFileHandle.getChannel().force(true);
-            LOG.info("Preallocated " + checkpointFile + " to " + checkpointFileHandle.length() + " for capacity "
+            logger.info("Preallocated " + checkpointFile + " to " + checkpointFileHandle.length() + " for capacity "
                     + capacity);
         }
         if (checkpointFile.length() != totalBytes) {
@@ -187,7 +187,7 @@ protected void backupCheckpoint(File backupDirectory) throws IOException {
                         + "this directory: "
                         + backupDirectory.toString() + " as backup?");
         if (!backupFile.createNewFile()) {
-            LOG.error("Could not create backup file. Backup of checkpoint will "
+            logger.error("Could not create backup file. Backup of checkpoint will "
                     + "not be used during replay even if checkpoint is bad.");
         }
     }
@@ -225,7 +225,7 @@ public static boolean restoreBackup(File checkpointDir, File backupDir) throws I
 
     @Override
     void beginCheckpoint() throws IOException {
-        LOG.info("Start checkpoint for " + checkpointFile + ", elements to sync = " + overwriteMap.size());
+        logger.info("Start checkpoint for " + checkpointFile + ", elements to sync = " + overwriteMap.size());
 
         if (shouldBackup) {
             int permits = backupCompletedSema.drainPermits();
@@ -250,7 +250,7 @@ void beginCheckpoint() throws IOException {
     void checkpoint() throws IOException {
 
         setLogWriteOrderID(WriteOrderOracle.next());
-        LOG.info("Updating checkpoint metadata: logWriteOrderID: "
+        logger.info("Updating checkpoint metadata: logWriteOrderID: "
                 + getLogWriteOrderID() + ", queueSize: " + getSize() + ", queueHead: "
                 + getHead());
         elementsBuffer.put(INDEX_WRITE_ORDER_ID, getLogWriteOrderID());
@@ -286,7 +286,7 @@ private void startBackupThread() {
                 checkpointBackUpExecutor,
                 "Expected the checkpoint backup exector to be non-null, "
                         + "but it is null. Checkpoint will not be backed up.");
-        LOG.info("Attempting to back up checkpoint.");
+        logger.info("Attempting to back up checkpoint.");
         checkpointBackUpExecutor.submit(new Runnable() {
 
             @Override
@@ -297,12 +297,12 @@ public void run() {
                 } catch (Throwable throwable) {
                     fileChannelCounter.incrementCheckpointBackupWriteErrorCount();
                     error = true;
-                    LOG.error("Backing up of checkpoint directory failed.", throwable);
+                    logger.error("Backing up of checkpoint directory failed.", throwable);
                 } finally {
                     backupCompletedSema.release();
                 }
                 if (!error) {
-                    LOG.info("Checkpoint backup completed.");
+                    logger.info("Checkpoint backup completed.");
                 }
             }
         });
@@ -314,15 +314,15 @@ void close() {
         try {
             checkpointFileHandle.close();
         } catch (IOException e) {
-            LOG.info("Error closing " + checkpointFile, e);
+            logger.info("Error closing " + checkpointFile, e);
         }
         if (checkpointBackUpExecutor != null && !checkpointBackUpExecutor.isShutdown()) {
             checkpointBackUpExecutor.shutdown();
             try {
                 // Wait till the executor dies.
                 while (!checkpointBackUpExecutor.awaitTermination(1, TimeUnit.SECONDS)) {}
             } catch (InterruptedException ex) {
-                LOG.warn("Interrupted while waiting for checkpoint backup to " + "complete");
+                logger.warn("Interrupted while waiting for checkpoint backup to " + "complete");
             }
         }
     }