Skip to content

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

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

Merged
ppkarwasz merged 4 commits into from
Jun 17, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5a25643
Migrate logging API from SLF4J to Log4j 2.x
ppkarwasz Jun 17, 2026
f873e14
Formatting
ppkarwasz Jun 17, 2026
b7825d3
Ban transitive logging implementations and bridges
ppkarwasz Jun 17, 2026
d7934cd
Pin SLF4J to 2.x and use the SLF4J 2 binding
ppkarwasz Jun 17, 2026
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
11 changes: 3 additions & 8 deletions flume-ng-auth/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>

Expand All @@ -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>
Expand Down
18 changes: 9 additions & 9 deletions flume-ng-auth/src/main/java/org/apache/flume/auth/KerberosAuthenticator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}

/*
Expand All @@ -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);
Expand All @@ -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,
Expand All @@ -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);
}
Expand Down
18 changes: 3 additions & 15 deletions flume-ng-channels/flume-file-channel/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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>

Expand Down
20 changes: 10 additions & 10 deletions ...s/flume-file-channel/src/main/java/org/apache/flume/channel/file/CheckpointRebuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand All @@ -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 {
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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;
}

Expand All @@ -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();
Expand Down Expand Up @@ -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"));
Expand All @@ -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.");
}
}
}
Expand Down
14 changes: 7 additions & 7 deletions ...le-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {}

Expand Down Expand Up @@ -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, "
Expand Down Expand Up @@ -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);
}
}
}
Expand All @@ -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();
Expand Down
24 changes: 12 additions & 12 deletions ...-file-channel/src/main/java/org/apache/flume/channel/file/EventQueueBackingStoreFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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.");
}
}
Expand Down Expand Up @@ -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();
Expand All @@ -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());
Expand Down Expand Up @@ -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
Expand All @@ -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.");
}
}
});
Expand All @@ -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");
}
}
}
Expand Down
Loading
Loading