You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@logging.apache.org by vy...@apache.org on 2020/12/28 22:18:44 UTC
[logging-log4j2] branch LOG4J2-2972 created (now dc75835)
This is an automated email from the ASF dual-hosted git repository.
vy pushed a change to branch LOG4J2-2972
in repository https://gitbox.apache.org/repos/asf/logging-log4j2.git.
at dc75835 LOG4J2-2972 Respawn AsyncAppender thread on failures.
This branch includes the following new commits:
new dc75835 LOG4J2-2972 Respawn AsyncAppender thread on failures.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[logging-log4j2] 01/01: LOG4J2-2972 Respawn AsyncAppender thread on
failures.
Posted by vy...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vy pushed a commit to branch LOG4J2-2972
in repository https://gitbox.apache.org/repos/asf/logging-log4j2.git
commit dc7583597953188973668b959ef6b16f97386f9e
Author: Volkan Yazici <vo...@gmail.com>
AuthorDate: Mon Dec 28 23:18:06 2020 +0100
LOG4J2-2972 Respawn AsyncAppender thread on failures.
---
.../org/apache/logging/log4j/util/SneakyThrow.java | 29 ---
.../logging/log4j/core/appender/AsyncAppender.java | 140 ++-----------
.../core/appender/AsyncAppenderEventForwarder.java | 232 +++++++++++++++++++++
.../logging/log4j/core/config/AppenderControl.java | 2 +-
.../async/AsyncAppenderExceptionHandlingTest.java | 9 +-
.../log4j/test/appender/FailOnceAppender.java | 15 +-
6 files changed, 270 insertions(+), 157 deletions(-)
diff --git a/log4j-api/src/test/java/org/apache/logging/log4j/util/SneakyThrow.java b/log4j-api/src/test/java/org/apache/logging/log4j/util/SneakyThrow.java
deleted file mode 100644
index e3beac5..0000000
--- a/log4j-api/src/test/java/org/apache/logging/log4j/util/SneakyThrow.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache license, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the license for the specific language governing permissions and
- * limitations under the license.
- */
-package org.apache.logging.log4j.util;
-
-public enum SneakyThrow {;
-
- /**
- * Throws any exception (including checked ones!) without defining it in the method signature.
- */
- @SuppressWarnings("unchecked")
- public static <E extends Throwable> void sneakyThrow(final Throwable throwable) throws E {
- throw (E) throwable;
- }
-
-}
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/AsyncAppender.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/AsyncAppender.java
index 0779ab3..757faa6 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/AsyncAppender.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/AsyncAppender.java
@@ -16,15 +16,6 @@
*/
package org.apache.logging.log4j.core.appender;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TransferQueue;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.logging.log4j.core.AbstractLogEvent;
import org.apache.logging.log4j.core.Appender;
import org.apache.logging.log4j.core.Core;
import org.apache.logging.log4j.core.Filter;
@@ -51,9 +42,17 @@ import org.apache.logging.log4j.core.config.plugins.PluginElement;
import org.apache.logging.log4j.core.config.plugins.validation.constraints.Required;
import org.apache.logging.log4j.core.filter.AbstractFilterable;
import org.apache.logging.log4j.core.impl.Log4jLogEvent;
-import org.apache.logging.log4j.core.util.Log4jThread;
+import org.apache.logging.log4j.core.util.Log4jThreadFactory;
import org.apache.logging.log4j.spi.AbstractLogger;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TransferQueue;
+
/**
* Appends to one or more Appenders asynchronously. You can configure an AsyncAppender with one or more Appenders and an
* Appender to append to if the queue is full. The AsyncAppender does not allow a filter to be specified on the Appender
@@ -63,11 +62,6 @@ import org.apache.logging.log4j.spi.AbstractLogger;
public final class AsyncAppender extends AbstractAppender {
private static final int DEFAULT_QUEUE_SIZE = 1024;
- private static final LogEvent SHUTDOWN_LOG_EVENT = new AbstractLogEvent() {
- private static final long serialVersionUID = -1761035149477086330L;
- };
-
- private static final AtomicLong THREAD_SEQUENCE = new AtomicLong(1);
private final BlockingQueue<LogEvent> queue;
private final int queueSize;
@@ -78,7 +72,7 @@ public final class AsyncAppender extends AbstractAppender {
private final String errorRef;
private final boolean includeLocation;
private AppenderControl errorAppender;
- private AsyncThread thread;
+ private AsyncAppenderEventForwarder forwarder;
private AsyncQueueFullPolicy asyncQueueFullPolicy;
private AsyncAppender(final String name, final Filter filter, final AppenderRef[] appenderRefs,
@@ -117,14 +111,16 @@ public final class AsyncAppender extends AbstractAppender {
}
}
if (appenders.size() > 0) {
- thread = new AsyncThread(appenders, queue);
- thread.setName("AsyncAppender-" + getName());
+ final ThreadFactory threadFactory =
+ new Log4jThreadFactory(getName(), true, Thread.NORM_PRIORITY);
+ forwarder = new AsyncAppenderEventForwarder(
+ errorAppender, appenders, queue, threadFactory);
} else if (errorRef == null) {
throw new ConfigurationException("No appenders are available for AsyncAppender " + getName());
}
asyncQueueFullPolicy = AsyncQueueFullPolicyFactory.create();
- thread.start();
+ forwarder.start();
super.start();
}
@@ -133,9 +129,8 @@ public final class AsyncAppender extends AbstractAppender {
setStopping();
super.stop(timeout, timeUnit, false);
LOGGER.trace("AsyncAppender stopping. Queue still has {} events.", queue.size());
- thread.shutdown();
try {
- thread.join(shutdownTimeout);
+ forwarder.stop(shutdownTimeout);
} catch (final InterruptedException ex) {
LOGGER.warn("Interrupted while stopping AsyncAppender {}", getName());
}
@@ -169,7 +164,7 @@ public final class AsyncAppender extends AbstractAppender {
logMessageInCurrentThread(logEvent);
} else {
// delegate to the event router (which may discard, enqueue and block, or log in current thread)
- final EventRoute route = asyncQueueFullPolicy.getRoute(thread.getId(), memento.getLevel());
+ final EventRoute route = asyncQueueFullPolicy.getRoute(forwarder.getActiveThreadId(), memento.getLevel());
route.logMessage(this, memento);
}
} else {
@@ -192,8 +187,7 @@ public final class AsyncAppender extends AbstractAppender {
*/
public void logMessageInCurrentThread(final LogEvent logEvent) {
logEvent.setEndOfBatch(queue.isEmpty());
- final boolean appendSuccessful = thread.callAppenders(logEvent);
- logToErrorAppenderIfNecessary(appendSuccessful, logEvent);
+ forwarder.forwardOne(logEvent);
}
/**
@@ -375,103 +369,6 @@ public final class AsyncAppender extends AbstractAppender {
}
/**
- * Thread that calls the Appenders.
- */
- private class AsyncThread extends Log4jThread {
-
- private volatile boolean shutdown;
- private final List<AppenderControl> appenders;
- private final BlockingQueue<LogEvent> queue;
-
- public AsyncThread(final List<AppenderControl> appenders, final BlockingQueue<LogEvent> queue) {
- super("AsyncAppender-" + THREAD_SEQUENCE.getAndIncrement());
- this.appenders = appenders;
- this.queue = queue;
- setDaemon(true);
- }
-
- @Override
- public void run() {
- while (!shutdown) {
- LogEvent event;
- try {
- event = queue.take();
- if (event == SHUTDOWN_LOG_EVENT) {
- shutdown = true;
- continue;
- }
- } catch (final InterruptedException ex) {
- break; // LOG4J2-830
- }
- event.setEndOfBatch(queue.isEmpty());
- final boolean success = callAppenders(event);
- if (!success && errorAppender != null) {
- try {
- errorAppender.callAppender(event);
- } catch (final Exception ex) {
- // Silently accept the error.
- }
- }
- }
- // Process any remaining items in the queue.
- LOGGER.trace("AsyncAppender.AsyncThread shutting down. Processing remaining {} queue events.",
- queue.size());
- int count = 0;
- int ignored = 0;
- while (!queue.isEmpty()) {
- try {
- final LogEvent event = queue.take();
- if (event instanceof Log4jLogEvent) {
- final Log4jLogEvent logEvent = (Log4jLogEvent) event;
- logEvent.setEndOfBatch(queue.isEmpty());
- callAppenders(logEvent);
- count++;
- } else {
- ignored++;
- LOGGER.trace("Ignoring event of class {}", event.getClass().getName());
- }
- } catch (final InterruptedException ex) {
- // May have been interrupted to shut down.
- // Here we ignore interrupts and try to process all remaining events.
- }
- }
- LOGGER.trace("AsyncAppender.AsyncThread stopped. Queue has {} events remaining. "
- + "Processed {} and ignored {} events since shutdown started.", queue.size(), count, ignored);
- }
-
- /**
- * Calls {@link AppenderControl#callAppender(LogEvent) callAppender} on all registered {@code AppenderControl}
- * objects, and returns {@code true} if at least one appender call was successful, {@code false} otherwise. Any
- * exceptions are silently ignored.
- *
- * @param event the event to forward to the registered appenders
- * @return {@code true} if at least one appender call succeeded, {@code false} otherwise
- */
- boolean callAppenders(final LogEvent event) {
- boolean success = false;
- for (final AppenderControl control : appenders) {
- try {
- control.callAppender(event);
- success = true;
- } catch (final Exception ex) {
- // If no appender is successful the error appender will get it.
- }
- }
- return success;
- }
-
- public void shutdown() {
- shutdown = true;
- if (queue.isEmpty()) {
- queue.offer(SHUTDOWN_LOG_EVENT);
- }
- if (getState() == State.TIMED_WAITING || getState() == State.WAITING) {
- this.interrupt(); // LOG4J2-1422: if underlying appender is stuck in wait/sleep/join/park call
- }
- }
- }
-
- /**
* Returns the names of the appenders that this asyncAppender delegates to as an array of Strings.
*
* @return the names of the sink appenders
@@ -530,4 +427,5 @@ public final class AsyncAppender extends AbstractAppender {
public int getQueueSize() {
return queue.size();
}
+
}
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/AsyncAppenderEventForwarder.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/AsyncAppenderEventForwarder.java
new file mode 100644
index 0000000..2c68cf3
--- /dev/null
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/AsyncAppenderEventForwarder.java
@@ -0,0 +1,232 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache license, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the license for the specific language governing permissions and
+ * limitations under the license.
+ */
+package org.apache.logging.log4j.core.appender;
+
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.config.AppenderControl;
+import org.apache.logging.log4j.status.StatusLogger;
+import org.apache.logging.log4j.util.Supplier;
+
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ThreadFactory;
+
+class AsyncAppenderEventForwarder {
+
+ private static final Logger LOGGER = StatusLogger.getLogger();
+
+ private final AppenderControl errorAppender;
+
+ private final List<AppenderControl> appenders;
+
+ private final BlockingQueue<LogEvent> queue;
+
+ private final ThreadFactory threadFactory;
+
+ private volatile boolean stopped;
+
+ private volatile Thread activeThread;
+
+ AsyncAppenderEventForwarder(
+ final AppenderControl errorAppender,
+ final List<AppenderControl> appenders,
+ final BlockingQueue<LogEvent> queue,
+ final ThreadFactory threadFactory) {
+ this.errorAppender = errorAppender;
+ this.appenders = appenders;
+ this.queue = queue;
+ this.threadFactory = threadFactory;
+ this.stopped = true;
+ this.activeThread = null;
+ }
+
+ synchronized void start() {
+ if (stopped) {
+ final Thread thread = createForwarder();
+ stopped = false;
+ activeThread = thread;
+ thread.start();
+ }
+ }
+
+ private Thread createForwarder() {
+
+ // Create the holder for the thread supplier. (We need this holder to
+ // avoid recursive calls to createForwarder() in the uncaught exception
+ // handler, since this might result in a stack overflow.)
+ @SuppressWarnings("unchecked")
+ final Supplier<Thread>[] threadSupplierRef = new Supplier[1];
+
+ // Create the uncaught exception handler, which respawns the forwarder
+ // upon unexpected termination.
+ final Thread.UncaughtExceptionHandler uncaughtExceptionHandler =
+ (final Thread ignored, final Throwable error) -> {
+
+ // In a synchronized block, determine if respawning should commence.
+ final Thread nextActiveThread;
+ synchronized (AsyncAppenderEventForwarder.this) {
+ if (stopped) {
+ nextActiveThread = null;
+ } else {
+ nextActiveThread = threadSupplierRef[0].get();
+ activeThread = nextActiveThread;
+ }
+ }
+
+ // Execute the result determined above — no synchronization
+ // is needed at this stage.
+ if (nextActiveThread == null) {
+ LOGGER.warn("forwarder has failed", error);
+ } else {
+ LOGGER.warn("respawning failed forwarder", error);
+ nextActiveThread.start();
+ }
+
+ };
+
+ // Create the thread supplier injecting the above created uncaught
+ // exception handler.
+ final Supplier<Thread> threadSupplier = () -> {
+ final Thread thread =
+ threadFactory.newThread(this::forwardAll);
+ thread.setUncaughtExceptionHandler(uncaughtExceptionHandler);
+ return thread;
+ };
+
+ // Fill in the holder.
+ threadSupplierRef[0] = threadSupplier;
+
+ // Return the initial thread.
+ return threadSupplierRef[0].get();
+
+ }
+
+ private void forwardAll() {
+ // Here we don't need to check against the "stopped" flag, since it is
+ // only used for determining if respawning should take place.
+ final Thread thread = Thread.currentThread();
+ while (!thread.isInterrupted()) {
+ LogEvent event;
+ try {
+ event = queue.take();
+ } catch (final InterruptedException ignored) {
+ // Restore the interrupted flag cleared when the exception is caught.
+ thread.interrupt();
+ break;
+ }
+ event.setEndOfBatch(queue.isEmpty());
+ forwardOne(event);
+ }
+ }
+
+ synchronized long getActiveThreadId() {
+ return stopped ? -1 : activeThread.getId();
+ }
+
+ void stop(final long timeoutMillis) throws InterruptedException {
+
+ // Disable respawning, if necessary.
+ final Thread lastActiveThread;
+ synchronized (this) {
+ if (stopped) {
+ return;
+ } else {
+ lastActiveThread = activeThread;
+ activeThread = null;
+ stopped = true;
+ }
+ }
+
+ // Put the termination sequence into a thread.
+ final Thread stopper = threadFactory.newThread(() -> {
+
+ // Update the thread name to reflect the purpose.
+ final Thread thread = Thread.currentThread();
+ final String threadName = String.format("%s-Stopper", thread.getName());
+ thread.setName(threadName);
+
+ // There is a slight chance that the last active forwarder thread is
+ // not started yet, wait for it to get picked up by a thread.
+ // Otherwise, interrupt+join will block.
+ // noinspection LoopConditionNotUpdatedInsideLoop, StatementWithEmptyBody
+ while (Thread.State.NEW.equals(lastActiveThread.getState()));
+
+ // Interrupt the last active forwarder.
+ lastActiveThread.interrupt();
+
+ // Forward any last remaining events.
+ forwardRemaining();
+
+ // Wait for the last active forwarder to stop.
+ try {
+ lastActiveThread.join();
+ } catch (final InterruptedException ignored) {
+ // Restore the interrupted flag cleared when the exception is caught.
+ thread.interrupt();
+ }
+
+ });
+
+ // Commence the termination sequence and wait at most for the given amount of time.
+ stopper.start();
+ stopper.join(timeoutMillis);
+
+ }
+
+ private void forwardRemaining() {
+ int eventCount = 0;
+ while (true) {
+ final LogEvent event = queue.poll();
+ if (event == null) {
+ break;
+ }
+ event.setEndOfBatch(queue.isEmpty());
+ forwardOne(event);
+ eventCount++;
+ }
+ LOGGER.trace("AsyncAppenderThreadTask processed the last {} remaining event(s).", eventCount);
+ }
+
+ void forwardOne(final LogEvent event) {
+
+ // Forward the event to all registered appenders.
+ boolean succeeded = false;
+ // noinspection ForLoopReplaceableByForEach (avoid iterator instantion)
+ for (int appenderIndex = 0; appenderIndex < appenders.size(); appenderIndex++) {
+ final AppenderControl control = appenders.get(appenderIndex);
+ try {
+ control.callAppender(event);
+ succeeded = true;
+ } catch (final Exception ignored) {
+ // If no appender is successful, the error appender will get it.
+ }
+ }
+
+ // Fallback to the error appender if none has succeeded so far.
+ if (!succeeded && errorAppender != null) {
+ try {
+ errorAppender.callAppender(event);
+ } catch (final Exception ignored) {
+ // If the error appender also fails, there is nothing further
+ // we can do about it.
+ }
+ }
+
+ }
+
+}
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/config/AppenderControl.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/config/AppenderControl.java
index c5a0d2a..58c1e57 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/config/AppenderControl.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/config/AppenderControl.java
@@ -156,7 +156,7 @@ public class AppenderControl extends AbstractFilterable {
appender.append(event);
} catch (final RuntimeException error) {
handleAppenderError(event, error);
- } catch (final Throwable error) {
+ } catch (final Exception error) {
handleAppenderError(event, new AppenderLoggingException(error));
}
}
diff --git a/log4j-core/src/test/java/org/apache/logging/log4j/core/async/AsyncAppenderExceptionHandlingTest.java b/log4j-core/src/test/java/org/apache/logging/log4j/core/async/AsyncAppenderExceptionHandlingTest.java
index 8318c81..164bdca 100644
--- a/log4j-core/src/test/java/org/apache/logging/log4j/core/async/AsyncAppenderExceptionHandlingTest.java
+++ b/log4j-core/src/test/java/org/apache/logging/log4j/core/async/AsyncAppenderExceptionHandlingTest.java
@@ -39,10 +39,10 @@ import java.util.stream.Collectors;
* Verifies {@link AsyncAppender} works after certain type of {@link Appender}
* failures.
* <p>
- * {@link AsyncAppender} thread is known to get killed due to
+ * {@code AsyncAppender} thread is known to get killed due to
* {@link AppenderControl} leaking exceptions in the past. This class is more
- * of an end-to-end test to verify that {@link AppenderControl} catches all kind
- * of {@link Throwable}s.
+ * of an end-to-end test to verify that {@code AsyncAppender} still works even
+ * if the background thread gets killed.
*/
class AsyncAppenderExceptionHandlingTest {
@@ -52,7 +52,8 @@ class AsyncAppenderExceptionHandlingTest {
FailOnceAppender.ThrowableClassName.LOGGING_EXCEPTION,
FailOnceAppender.ThrowableClassName.EXCEPTION,
FailOnceAppender.ThrowableClassName.ERROR,
- FailOnceAppender.ThrowableClassName.THROWABLE
+ FailOnceAppender.ThrowableClassName.THROWABLE,
+ FailOnceAppender.ThrowableClassName.THREAD_DEATH
})
void AsyncAppender_should_not_stop_on_appender_failures(String throwableClassName) {
diff --git a/log4j-core/src/test/java/org/apache/logging/log4j/test/appender/FailOnceAppender.java b/log4j-core/src/test/java/org/apache/logging/log4j/test/appender/FailOnceAppender.java
index 1f5c932..ae6db66 100644
--- a/log4j-core/src/test/java/org/apache/logging/log4j/test/appender/FailOnceAppender.java
+++ b/log4j-core/src/test/java/org/apache/logging/log4j/test/appender/FailOnceAppender.java
@@ -25,7 +25,7 @@ import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginFactory;
import org.apache.logging.log4j.core.config.plugins.validation.constraints.Required;
-import org.apache.logging.log4j.util.SneakyThrow;
+import org.apache.logging.log4j.core.util.Throwables;
import java.util.ArrayList;
import java.util.List;
@@ -53,7 +53,7 @@ public class FailOnceAppender extends AbstractAppender {
if (!failed) {
failed = true;
Throwable throwable = throwableSupplier.get();
- SneakyThrow.sneakyThrow(throwable);
+ Throwables.rethrow(throwable);
}
events.add(event);
}
@@ -95,11 +95,20 @@ public class FailOnceAppender extends AbstractAppender {
case ThrowableClassName.EXCEPTION: return () -> new Exception(message);
case ThrowableClassName.ERROR: return () -> new Error(message);
case ThrowableClassName.THROWABLE: return () -> new Throwable(message);
+ case ThrowableClassName.THREAD_DEATH: return () -> {
+ stopCurrentThread();
+ throw new IllegalStateException("should not have reached here");
+ };
default: throw new IllegalArgumentException("unknown throwable class name: " + throwableClassName);
}
}
+ @SuppressWarnings("deprecation")
+ private static void stopCurrentThread() {
+ Thread.currentThread().stop();
+ }
+
public enum ThrowableClassName {;
public static final String RUNTIME_EXCEPTION = "RuntimeException";
@@ -112,6 +121,8 @@ public class FailOnceAppender extends AbstractAppender {
public static final String THROWABLE = "Throwable";
+ public static final String THREAD_DEATH = "ThreadDeath";
+
}
}