diff --git a/rollbar-android/src/main/java/com/rollbar/android/Rollbar.java b/rollbar-android/src/main/java/com/rollbar/android/Rollbar.java index 2200fbdf..ddd70e46 100644 --- a/rollbar-android/src/main/java/com/rollbar/android/Rollbar.java +++ b/rollbar-android/src/main/java/com/rollbar/android/Rollbar.java @@ -486,7 +486,11 @@ public Rollbar(Context context, String accessToken, String environment, .notifier(new NotifierProvider(context)) .environment(environment == null ? DEFAULT_ENVIRONMENT : environment) .sender(sender) - .handleUncaughtErrors(false); // Use the global handler, not the default per thread one. + .handleUncaughtErrors(false) // Use the global handler, not the default per thread one. + // The DiskQueue above already persists payloads before the process dies, and they are + // transmitted on the next app start. Flushing at shutdown would attempt blocking network + // I/O while the app is being torn down, with nothing to gain. + .flushOnShutdown(false); Config config; if (configProvider != null) { diff --git a/rollbar-java/src/main/java/com/rollbar/notifier/Rollbar.java b/rollbar-java/src/main/java/com/rollbar/notifier/Rollbar.java index b9265745..dc4945a4 100644 --- a/rollbar-java/src/main/java/com/rollbar/notifier/Rollbar.java +++ b/rollbar-java/src/main/java/com/rollbar/notifier/Rollbar.java @@ -5,6 +5,9 @@ import com.rollbar.notifier.config.Config; import com.rollbar.notifier.config.ConfigBuilder; import com.rollbar.notifier.config.ConfigProvider; +import com.rollbar.notifier.provider.Provider; +import com.rollbar.notifier.sender.Sender; +import com.rollbar.notifier.sender.SenderShutdownHooks; import com.rollbar.notifier.uncaughtexception.RollbarUncaughtExceptionHandler; import com.rollbar.notifier.util.BodyFactory; import com.rollbar.notifier.util.ObjectsUtils; @@ -26,6 +29,10 @@ public class Rollbar extends RollbarBase { private static volatile Rollbar notifier; + private final Object shutdownHookLock = new Object(); + + private Thread shutdownHook; + /** * Constructor. * @@ -41,9 +48,95 @@ public Rollbar(Config config) { if (config.handleUncaughtErrors()) { this.handleUncaughtErrors(); } + if (config.flushOnShutdown()) { + this.registerShutdownHook(config); + } processAppPackages(config); } + /** + * Registers a JVM shutdown hook that flushes buffered payloads before the process exits. + * + *

+ * Mirrors {@link #handleUncaughtErrors()} in being driven from the configuration at + * construction time, so replacing the configuration later via + * {@link #configure(ConfigProvider)} does not add or remove the hook. + *

+ */ + private void registerShutdownHook(Config config) { + if (config.sender() == null) { + return; + } + + Thread hook = SenderShutdownHooks.create(new CurrentSenderProvider(), + config.shutdownTimeoutMillis()); + try { + Runtime.getRuntime().addShutdownHook(hook); + synchronized (shutdownHookLock) { + this.shutdownHook = hook; + } + LOGGER.debug("Registered the Rollbar shutdown hook."); + } catch (IllegalStateException e) { + // The JVM is already shutting down, so there is nothing left to flush later. + LOGGER.debug("The JVM is already shutting down, the Rollbar shutdown hook was not " + + "registered."); + } catch (SecurityException e) { + LOGGER.warn("No permission to register the Rollbar shutdown hook. Payloads buffered when " + + "the JVM exits will not be sent.", e); + } + } + + /** + * Resolves the sender from the configuration in force when the hook actually runs, so that a + * configuration replaced through {@link #configure(ConfigProvider)} after the hook was + * registered does not leave a stale sender being flushed. + */ + private class CurrentSenderProvider implements Provider { + @Override + public Sender provide() { + configReadLock.lock(); + try { + return config.sender(); + } finally { + configReadLock.unlock(); + } + } + } + + /** + * The registered shutdown hook, or null if none was registered. Visible for testing. + */ + Thread shutdownHook() { + synchronized (shutdownHookLock) { + return this.shutdownHook; + } + } + + /** + * Removes the shutdown hook, if one was registered. Called when the notifier is closed + * explicitly, both to avoid flushing twice and to let the hook be garbage collected. + */ + private void unregisterShutdownHook() { + Thread hook; + synchronized (shutdownHookLock) { + hook = this.shutdownHook; + this.shutdownHook = null; + } + + if (hook == null) { + return; + } + + try { + Runtime.getRuntime().removeShutdownHook(hook); + } catch (IllegalStateException e) { + // Shutdown is already in progress, the hook may be running right now. Nothing to do. + LOGGER.debug("The JVM is already shutting down, the Rollbar shutdown hook was not removed."); + } catch (SecurityException e) { + LOGGER.debug("No permission to remove the Rollbar shutdown hook."); + } + } + /** * Method to initialize the library managed notifier instance. * @@ -651,7 +744,16 @@ public void log(ThrowableWrapper error, Map custom, String descr } } + /** + * Close the notifier and its sender, unregistering the shutdown hook first. + * + * @param wait whether to wait for the sender to send any buffered payloads before closing. + * @throws Exception if closing the sender fails. + */ public void close(boolean wait) throws Exception { + // Dropped first: the notifier is being shut down explicitly, so the hook would either flush a + // sender that is already closed or duplicate the work being done here. + unregisterShutdownHook(); this.config.sender().close(wait); } diff --git a/rollbar-java/src/main/java/com/rollbar/notifier/config/Config.java b/rollbar-java/src/main/java/com/rollbar/notifier/config/Config.java index 4257bb7c..b154f7cc 100644 --- a/rollbar-java/src/main/java/com/rollbar/notifier/config/Config.java +++ b/rollbar-java/src/main/java/com/rollbar/notifier/config/Config.java @@ -23,4 +23,45 @@ public interface Config extends CommonConfig { * @return the proxy. */ Proxy proxy(); + + /** + *

+ * If set to true (the default), a JVM shutdown hook is registered that flushes any payloads + * still buffered in the {@link Sender sender} before the process exits. + *

+ *

+ * The default sender buffers payloads in memory and drains them on a background daemon thread + * every few seconds, so without this hook any occurrence captured shortly before the JVM + * terminates is discarded. That includes short-lived processes and, more importantly, + * {@code SIGTERM} during a rolling deploy or container eviction. + *

+ *

+ * Set to false if the application manages the notifier lifecycle itself, for example by + * calling {@link Rollbar#close(boolean)} explicitly, or when payloads are already persisted by + * a durable queue. + *

+ * + * @return true to flush buffered payloads on JVM shutdown, false otherwise. + */ + default boolean flushOnShutdown() { + return true; + } + + /** + *

+ * The maximum time, in milliseconds, that the shutdown hook installed by + * {@link #flushOnShutdown()} will spend flushing buffered payloads before letting the JVM + * continue to exit. Ignored when {@link #flushOnShutdown()} is false. + *

+ *

+ * This is a hard bound and not a target: shutdown always proceeds once it elapses, even if + * payloads remain unsent. It exists because a flush performs blocking network I/O, and an + * unbounded one would be able to stall JVM shutdown indefinitely. + *

+ * + * @return the shutdown flush timeout in milliseconds. + */ + default long shutdownTimeoutMillis() { + return 2000L; + } } diff --git a/rollbar-java/src/main/java/com/rollbar/notifier/config/ConfigBuilder.java b/rollbar-java/src/main/java/com/rollbar/notifier/config/ConfigBuilder.java index e2fb894d..2fc772cc 100644 --- a/rollbar-java/src/main/java/com/rollbar/notifier/config/ConfigBuilder.java +++ b/rollbar-java/src/main/java/com/rollbar/notifier/config/ConfigBuilder.java @@ -36,6 +36,17 @@ */ public class ConfigBuilder { + /** + * Whether a JVM shutdown hook that flushes buffered payloads is registered by default. + */ + public static final boolean DEFAULT_FLUSH_ON_SHUTDOWN = true; + + /** + * The default hard bound, in milliseconds, on the shutdown flush. Kept short because it delays + * JVM exit, and the flush only has to drain what the background sender has not sent yet. + */ + public static final long DEFAULT_SHUTDOWN_TIMEOUT_MILLIS = 2000L; + protected String accessToken; protected String endpoint; @@ -92,6 +103,10 @@ public class ConfigBuilder { protected boolean compressPayload; + protected boolean flushOnShutdown; + + protected long shutdownTimeoutMillis; + protected List redactedKeys; protected boolean useDefaultRedactedKeys = true; @@ -113,6 +128,8 @@ protected ConfigBuilder(String accessToken) { this.handleUncaughtErrors = true; this.enabled = true; this.compressPayload = true; + this.flushOnShutdown = DEFAULT_FLUSH_ON_SHUTDOWN; + this.shutdownTimeoutMillis = DEFAULT_SHUTDOWN_TIMEOUT_MILLIS; this.defaultLevels = new DefaultLevels(); } @@ -149,6 +166,8 @@ private ConfigBuilder(Config config) { this.defaultLevels = new DefaultLevels(config); this.truncateLargePayloads = config.truncateLargePayloads(); this.compressPayload = config.compressPayload(); + this.flushOnShutdown = config.flushOnShutdown(); + this.shutdownTimeoutMillis = config.shutdownTimeoutMillis(); this.maximumTelemetryData = config.maximumTelemetryData(); this.telemetryEventTracker = config.telemetryEventTracker(); this.redactedKeys = config.redactedKeys(); @@ -508,6 +527,50 @@ public ConfigBuilder compressPayload(boolean compress) { return this; } + /** + *

+ * If set to true (the default), a JVM shutdown hook is registered that flushes any payloads + * still buffered in the sender before the process exits. + *

+ *

+ * The default sender buffers payloads in memory and drains them on a background daemon thread + * every few seconds, so without this hook an occurrence captured shortly before the JVM + * terminates is discarded. That covers short-lived processes as well as {@code SIGTERM} during + * a rolling deploy or container eviction. + *

+ *

+ * Set to false when the application manages the notifier lifecycle itself, for example by + * calling {@link com.rollbar.notifier.Rollbar#close(boolean)}, or when a durable queue already + * persists payloads across restarts. + *

+ * + * @param flushOnShutdown true to flush buffered payloads on JVM shutdown. + * @return the builder instance. + */ + public ConfigBuilder flushOnShutdown(boolean flushOnShutdown) { + this.flushOnShutdown = flushOnShutdown; + return this; + } + + /** + *

+ * The maximum time, in milliseconds, that the shutdown hook will spend flushing buffered + * payloads before letting the JVM continue to exit. Default: 2000. Ignored when + * {@link #flushOnShutdown(boolean)} is false. + *

+ *

+ * This is a hard bound rather than a target, and shutdown proceeds once it elapses even if + * payloads remain unsent. Values of zero or less start the flush without waiting for it. + *

+ * + * @param shutdownTimeoutMillis the shutdown flush timeout in milliseconds. + * @return the builder instance. + */ + public ConfigBuilder shutdownTimeoutMillis(long shutdownTimeoutMillis) { + this.shutdownTimeoutMillis = shutdownTimeoutMillis; + return this; + } + /** *

* Maximum Telemetry events sent in a payload, only for the default TelemetryEventTracker, if @@ -672,6 +735,10 @@ private static class ConfigImpl implements Config { private final boolean compressPayload; + private final boolean flushOnShutdown; + + private final long shutdownTimeoutMillis; + private final int maximumTelemetryData; private final TelemetryEventTracker telemetryEventTracker; @@ -715,6 +782,8 @@ private static class ConfigImpl implements Config { this.defaultLevels = builder.defaultLevels; this.truncateLargePayloads = builder.truncateLargePayloads; this.compressPayload = builder.compressPayload; + this.flushOnShutdown = builder.flushOnShutdown; + this.shutdownTimeoutMillis = builder.shutdownTimeoutMillis; this.maximumTelemetryData = builder.maximumTelemetryData; this.telemetryEventTracker = builder.telemetryEventTracker; this.redactedKeys = builder.redactedKeys != null @@ -874,6 +943,16 @@ public boolean compressPayload() { return this.compressPayload; } + @Override + public boolean flushOnShutdown() { + return this.flushOnShutdown; + } + + @Override + public long shutdownTimeoutMillis() { + return this.shutdownTimeoutMillis; + } + @Override public int maximumTelemetryData() { return this.maximumTelemetryData; diff --git a/rollbar-java/src/main/java/com/rollbar/notifier/sender/BufferedSender.java b/rollbar-java/src/main/java/com/rollbar/notifier/sender/BufferedSender.java index 8a0114ab..fbb728a3 100644 --- a/rollbar-java/src/main/java/com/rollbar/notifier/sender/BufferedSender.java +++ b/rollbar-java/src/main/java/com/rollbar/notifier/sender/BufferedSender.java @@ -10,7 +10,10 @@ import java.util.List; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; @@ -48,6 +51,8 @@ public class BufferedSender implements Sender { private SendTask sendTask; + private volatile boolean sendImmediately; + private static Logger LOGGER = LoggerFactory.getLogger(BufferedSender.class); BufferedSender(Builder builder) { @@ -98,12 +103,62 @@ public Sender sender() { @Override public void send(Payload payload) { + if (!enqueue(payload)) { + return; + } + + if (sendImmediately) { + try { + this.executorService.execute(this.sendTask); + } catch (RejectedExecutionException e) { + // Closed in the meantime. The payload stays queued, as it would for any closed sender. + } + } + } + + private boolean enqueue(Payload payload) { try { // If the queue is full it will raise an exception and it will be notified. queue.add(payload); + return true; } catch (Exception e) { notifyError(payload, new SenderException(e)); + return false; + } + } + + /** + * Sends the payloads still buffered, waiting for a send already in progress, and from then on + * sends each new payload as soon as it arrives instead of on the next scheduled run. + * + *

+ * Meant for JVM shutdown. Unlike {@link #close(boolean) close(true)}, this leaves the sender + * open, because shutdown hooks run concurrently: the application may still be logging errors, + * for example from requests that fail while a web server drains, long after this returns. The + * background thread is a daemon, so keeping it running never keeps the JVM alive. + *

+ * + *

+ * This method does not time out. Callers that need a bound, such as the shutdown hook, impose + * it themselves. + *

+ * + * @throws InterruptedException if interrupted while waiting for the payloads to be sent. + * @throws ExecutionException if sending the payloads fails. + */ + void flushForShutdown() throws InterruptedException, ExecutionException { + this.sendImmediately = true; + + Future flush; + try { + // The executor runs one task at a time, so this only starts once a run already in progress + // has finished. That run may have taken payloads off the queue that it is still sending. + flush = this.executorService.submit(this::flushQueue); + } catch (RejectedExecutionException e) { + // Already closed, and closing flushes on its own. + return; } + flush.get(); } @Override @@ -127,11 +182,23 @@ public void close() throws IOException { @Override public void close(boolean wait) throws Exception { - if (wait) { - this.flushQueue(); + if (!wait) { + this.close(); + return; } - this.close(); + try { + // Stop scheduling further runs and wait for one already in progress. That run may have + // taken payloads off the queue that it is still sending, so the queue being empty does not + // mean everything has been sent. Callers that need a bound on this, such as the shutdown + // hook, impose it themselves. + this.executorService.shutdown(); + this.executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + // Drain what is left, including payloads the in-flight run re-queued for retry. + this.flushQueue(); + } finally { + this.close(); + } } private void notifyError(Payload payload, Exception e) { @@ -142,7 +209,11 @@ private void notifyError(Payload payload, Exception e) { private void flushQueue() { while (queue.size() > 0) { - this.sendTask.run(); + if (this.sendTask.sendBatch() == 0) { + // Nothing was taken off the queue, which is what happens while sending is suspended. + // Stop, rather than spin for as long as the suspension lasts. + break; + } } } @@ -267,6 +338,15 @@ public SendTask(int batchSize, Queue queue, Sender sender, @Override public void run() { + sendBatch(); + } + + /** + * Sends up to one batch of payloads. + * + * @return the number of payloads taken off the queue, whether or not sending them succeeded. + */ + int sendBatch() { Payload payload = null; int numberOfSent = 0; @@ -302,6 +382,7 @@ public void run() { // This could be OOM, stack overflow, etc... So we can't call the listeners. Hopefully // logging still works, that's all we can do. } + return numberOfSent; } private Payload getItemFromQueue() { @@ -358,7 +439,10 @@ private void apply(Payload payload, SenderFailureStrategy.PayloadAction action) if (tooManySendAttempts(payload)) { LOGGER.warn("Discarding payload after " + payload.getSendAttemptCount() + " attempts"); } else { - send(payload); + // Not send(): after flushForShutdown() that would start a run right away, and a + // network outage would then burn through every attempt back to back. A retry waits + // for the next scheduled run instead, as it always has. + enqueue(payload); } break; default: diff --git a/rollbar-java/src/main/java/com/rollbar/notifier/sender/SenderShutdownHook.java b/rollbar-java/src/main/java/com/rollbar/notifier/sender/SenderShutdownHook.java new file mode 100644 index 00000000..b5acfbae --- /dev/null +++ b/rollbar-java/src/main/java/com/rollbar/notifier/sender/SenderShutdownHook.java @@ -0,0 +1,111 @@ +package com.rollbar.notifier.sender; + +import com.rollbar.notifier.provider.Provider; +import com.rollbar.notifier.util.ObjectsUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A JVM shutdown hook that flushes the payloads still buffered in a {@link Sender sender}. + * + *

+ * A {@link BufferedSender} is flushed without being closed, and keeps sending afterwards. The JVM + * runs shutdown hooks concurrently, so other hooks, such as a web server draining in-flight + * requests, may go on logging errors long after this one finishes. Other senders are closed with + * {@link Sender#close(boolean) close(true)}. + *

+ * + *

+ * The hook is bounded by a timeout. Flushing performs blocking network I/O, and the sender does + * not impose a socket timeout of its own, so an unbounded flush would be able to stall JVM + * shutdown for as long as the remote end kept the connection open. Instead the flush runs on a + * separate daemon thread and this hook waits for it for at most {@code timeoutMillis}; once that + * elapses shutdown continues regardless, and the daemon thread is torn down with the JVM. + *

+ * + *

+ * Package-private so that it can rely on {@link BufferedSender} internals without making them + * public. Created through {@link SenderShutdownHooks}. + *

+ */ +class SenderShutdownHook extends Thread { + + private static final Logger LOGGER = LoggerFactory.getLogger(SenderShutdownHook.class); + + private static final String HOOK_THREAD_NAME = "rollbar-shutdown-hook"; + + private static final String FLUSH_THREAD_NAME = "rollbar-shutdown-flush"; + + private final Provider senderProvider; + + private final long timeoutMillis; + + /** + * Constructor. + * + * @param senderProvider supplies the sender to flush, resolved when the hook runs rather than + * when it is registered, so that a sender replaced in the meantime through + * {@link com.rollbar.notifier.Rollbar#configure} is the one flushed. Not + * null, though it may return null. + * @param timeoutMillis the maximum time to spend flushing, in milliseconds. Values of zero or + * less disable waiting: the flush is still started, but shutdown does not + * block on it. + */ + SenderShutdownHook(Provider senderProvider, long timeoutMillis) { + super(HOOK_THREAD_NAME); + ObjectsUtils.requireNonNull(senderProvider, "The sender provider can not be null"); + this.senderProvider = senderProvider; + this.timeoutMillis = timeoutMillis; + } + + @Override + public void run() { + Thread flusher = new Thread(new FlushTask(), FLUSH_THREAD_NAME); + // Daemon, so that failing to finish within the timeout can never keep the JVM alive. + flusher.setDaemon(true); + flusher.start(); + + if (timeoutMillis <= 0) { + return; + } + + try { + flusher.join(timeoutMillis); + } catch (InterruptedException e) { + // Preserve the interrupt for whatever else is running during shutdown. + Thread.currentThread().interrupt(); + return; + } + + if (flusher.isAlive()) { + LOGGER.warn("Timed out after {} ms flushing Rollbar payloads on shutdown. Some occurrences " + + "may not have been sent.", timeoutMillis); + } + } + + private class FlushTask implements Runnable { + @Override + public void run() { + try { + Sender sender = senderProvider.provide(); + if (sender == null) { + return; + } + if (sender instanceof BufferedSender) { + // Not close(true): shutdown hooks run concurrently, so the application may keep logging + // after this returns, and a closed sender would leave those payloads queued forever. + ((BufferedSender) sender).flushForShutdown(); + } else { + sender.close(true); + } + } catch (Exception e) { + LOGGER.error("Error flushing Rollbar payloads on shutdown.", e); + } catch (Throwable e) { + // The JVM is on its way down and this thread has no other handler. Swallowing here keeps + // an error in the notifier from surfacing as a confusing failure during shutdown. + LOGGER.error("Error flushing Rollbar payloads on shutdown.", e); + } + } + } +} diff --git a/rollbar-java/src/main/java/com/rollbar/notifier/sender/SenderShutdownHooks.java b/rollbar-java/src/main/java/com/rollbar/notifier/sender/SenderShutdownHooks.java new file mode 100644 index 00000000..c2a2730f --- /dev/null +++ b/rollbar-java/src/main/java/com/rollbar/notifier/sender/SenderShutdownHooks.java @@ -0,0 +1,32 @@ +package com.rollbar.notifier.sender; + +import com.rollbar.notifier.provider.Provider; + +/** + * Creates the JVM shutdown hook that flushes the payloads still buffered in a sender. + * + *

+ * Used by {@link com.rollbar.notifier.Rollbar} to install the hook enabled through + * {@link com.rollbar.notifier.config.Config#flushOnShutdown()}. It only returns a plain + * {@link Thread}, so that how the hook flushes each kind of sender stays internal to this package. + *

+ */ +public final class SenderShutdownHooks { + + private SenderShutdownHooks() { + } + + /** + * Creates a shutdown hook, ready to be passed to {@link Runtime#addShutdownHook(Thread)}. + * + * @param senderProvider supplies the sender to flush, resolved when the hook runs rather than + * when it is created. Not null, though it may return null. + * @param timeoutMillis the maximum time to spend flushing, in milliseconds. Values of zero or + * less disable waiting: the flush is still started, but shutdown does not + * block on it. + * @return the shutdown hook. + */ + public static Thread create(Provider senderProvider, long timeoutMillis) { + return new SenderShutdownHook(senderProvider, timeoutMillis); + } +} diff --git a/rollbar-java/src/test/java/com/rollbar/notifier/RollbarShutdownHookTest.java b/rollbar-java/src/test/java/com/rollbar/notifier/RollbarShutdownHookTest.java new file mode 100644 index 00000000..3f6b874d --- /dev/null +++ b/rollbar-java/src/test/java/com/rollbar/notifier/RollbarShutdownHookTest.java @@ -0,0 +1,99 @@ +package com.rollbar.notifier; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.core.Is.is; +import static org.hamcrest.core.IsNull.notNullValue; +import static org.hamcrest.core.IsNull.nullValue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import com.rollbar.notifier.config.Config; +import com.rollbar.notifier.config.ConfigBuilder; +import com.rollbar.notifier.sender.Sender; + +import org.junit.Test; + +/** + * Covers the registration side of the shutdown hook. The flushing behaviour itself is covered by + * {@link com.rollbar.notifier.sender.SenderShutdownHookTest}. + */ +public class RollbarShutdownHookTest { + + @Test + public void shouldRegisterTheShutdownHookByDefault() throws Exception { + Rollbar rollbar = new Rollbar(configWith(mock(Sender.class), true)); + try { + assertThat(rollbar.shutdownHook(), is(notNullValue())); + } finally { + rollbar.close(false); + } + } + + @Test + public void shouldNotRegisterTheShutdownHookWhenDisabled() throws Exception { + Rollbar rollbar = new Rollbar(configWith(mock(Sender.class), false)); + try { + assertThat(rollbar.shutdownHook(), is(nullValue())); + } finally { + rollbar.close(false); + } + } + + @Test + public void shouldRemoveTheShutdownHookOnClose() throws Exception { + Rollbar rollbar = new Rollbar(configWith(mock(Sender.class), true)); + Thread hook = rollbar.shutdownHook(); + assertThat(hook, is(notNullValue())); + + rollbar.close(false); + + assertThat(rollbar.shutdownHook(), is(nullValue())); + // Idempotent: the hook is really gone from the JVM registry, not just from our field. + assertThat(Runtime.getRuntime().removeShutdownHook(hook), is(false)); + } + + /** + * Closing explicitly must flush exactly once. If the hook were left registered it would run a + * second flush against an already closed sender while the JVM was exiting. + */ + @Test + public void shouldFlushOnlyOnceWhenClosedExplicitly() throws Exception { + Sender sender = mock(Sender.class); + Rollbar rollbar = new Rollbar(configWith(sender, true)); + + rollbar.close(true); + + verify(sender).close(true); + assertThat(rollbar.shutdownHook(), is(nullValue())); + } + + @Test + public void shouldDefaultToFlushingOnShutdownWithATwoSecondBound() { + Config config = ConfigBuilder.withAccessToken("token").build(); + + assertThat(config.flushOnShutdown(), is(true)); + assertThat(config.shutdownTimeoutMillis(), is(2000L)); + } + + @Test + public void shouldCarryTheSettingsOverWhenRebuildingFromAnExistingConfig() { + Config original = ConfigBuilder.withAccessToken("token") + .flushOnShutdown(false) + .shutdownTimeoutMillis(750L) + .build(); + + Config copy = ConfigBuilder.withConfig(original).build(); + + assertThat(copy.flushOnShutdown(), is(false)); + assertThat(copy.shutdownTimeoutMillis(), is(750L)); + } + + private Config configWith(Sender sender, boolean flushOnShutdown) { + return ConfigBuilder.withAccessToken("token") + .sender(sender) + .handleUncaughtErrors(false) + .flushOnShutdown(flushOnShutdown) + .shutdownTimeoutMillis(100L) + .build(); + } +} diff --git a/rollbar-java/src/test/java/com/rollbar/notifier/sender/BufferedSenderTest.java b/rollbar-java/src/test/java/com/rollbar/notifier/sender/BufferedSenderTest.java index ccee60e0..3412cb97 100644 --- a/rollbar-java/src/test/java/com/rollbar/notifier/sender/BufferedSenderTest.java +++ b/rollbar-java/src/test/java/com/rollbar/notifier/sender/BufferedSenderTest.java @@ -19,8 +19,12 @@ import java.util.ArrayList; import java.util.List; import java.util.Queue; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; @@ -30,6 +34,7 @@ import org.junit.Rule; import org.junit.Test; import org.mockito.ArgumentCaptor; +import org.mockito.InOrder; import org.mockito.Mock; import org.mockito.invocation.InvocationOnMock; import org.mockito.junit.MockitoJUnit; @@ -129,13 +134,63 @@ public void shouldCloseWaiting() throws Exception { assertThat(queue.size(), is(0)); - verify(executorService).shutdown(); + InOrder inOrder = inOrder(executorService, sender); + inOrder.verify(executorService).shutdown(); + inOrder.verify(executorService).awaitTermination(anyLong(), any(TimeUnit.class)); + inOrder.verify(sender).send(payload1); verify(sender).send(payload1); verify(sender).send(payload2); verify(sender).close(); } + @Test + public void shouldCloseWaitingForInFlightSend() throws Exception { + CountDownLatch sendStarted = new CountDownLatch(1); + CountDownLatch releaseSend = new CountDownLatch(1); + AtomicBoolean sendFinished = new AtomicBoolean(false); + Payload payload = mock(Payload.class); + + doAnswer(invocation -> { + sendStarted.countDown(); + releaseSend.await(); + sendFinished.set(true); + return null; + }).when(sender).send(payload); + + Queue queue = new ConcurrentLinkedQueue<>(); + queue.add(payload); + + sut = new BufferedSender(new BufferedSender.Builder() + .queue(queue) + .sender(sender) + .initialFlushDelay(0), + Executors.newSingleThreadScheduledExecutor(new SenderThreadFactory())); + + // The scheduled run has taken the payload off the queue and is still sending it. + assertThat(sendStarted.await(5, TimeUnit.SECONDS), is(true)); + assertThat(queue.size(), is(0)); + + Thread closer = new Thread(() -> { + try { + sut.close(true); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + closer.start(); + + closer.join(200); + assertThat(closer.isAlive(), is(true)); + + releaseSend.countDown(); + closer.join(5000); + + assertThat(closer.isAlive(), is(false)); + assertThat(sendFinished.get(), is(true)); + verify(sender).close(); + } + @Test public void shouldCloseWithoutWaiting() throws Exception { Payload payload1 = mock(Payload.class); @@ -160,6 +215,183 @@ public void shouldCloseWithoutWaiting() throws Exception { verify(sender).close(); } + @Test(timeout = 5000) + public void shouldStopFlushingOnCloseWhileSendingIsSuspended() throws Exception { + Payload payload = mock(Payload.class); + Queue queue = new ConcurrentLinkedQueue<>(); + queue.add(payload); + + SenderFailureStrategy strategy = mock(SenderFailureStrategy.class); + when(strategy.isSendingSuspended()).thenReturn(true); + + sut = new BufferedSender(new BufferedSender.Builder() + .queue(queue) + .sender(sender) + .senderFailureStrategy(strategy), + executorService); + + sut.close(true); + + assertThat(queue.size(), is(1)); + verify(sender, never()).send(payload); + verify(sender).close(); + } + + @Test + public void shouldFlushForShutdownWithoutClosing() throws Exception { + Payload payload1 = mock(Payload.class); + Payload payload2 = mock(Payload.class); + + Queue queue = new ConcurrentLinkedQueue<>(); + queue.addAll(asList(payload1, payload2)); + + sut = bufferedSenderWithRealExecutor(queue); + + sut.flushForShutdown(); + + assertThat(queue.size(), is(0)); + verify(sender).send(payload1); + verify(sender).send(payload2); + verify(sender, never()).close(); + + sut.close(); + } + + @Test + public void shouldFlushForShutdownWaitingForInFlightSend() throws Exception { + CountDownLatch sendStarted = new CountDownLatch(1); + CountDownLatch releaseSend = new CountDownLatch(1); + AtomicBoolean sendFinished = new AtomicBoolean(false); + Payload inFlight = mock(Payload.class); + final Payload queued = mock(Payload.class); + + doAnswer(invocation -> { + sendStarted.countDown(); + releaseSend.await(); + sendFinished.set(true); + return null; + }).when(sender).send(inFlight); + + Queue queue = new ConcurrentLinkedQueue<>(); + queue.add(inFlight); + + sut = new BufferedSender(new BufferedSender.Builder() + .queue(queue) + .sender(sender) + .initialFlushDelay(0) + .flushFreq(TimeUnit.MINUTES.toMillis(1)), + Executors.newSingleThreadScheduledExecutor(new SenderThreadFactory())); + + // The scheduled run has taken the payload off the queue and is still sending it. + assertThat(sendStarted.await(5, TimeUnit.SECONDS), is(true)); + queue.add(queued); + + Thread flusher = new Thread(() -> { + try { + sut.flushForShutdown(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + flusher.start(); + + flusher.join(200); + assertThat(flusher.isAlive(), is(true)); + + releaseSend.countDown(); + flusher.join(5000); + + assertThat(flusher.isAlive(), is(false)); + assertThat(sendFinished.get(), is(true)); + verify(sender).send(queued); + verify(sender, never()).close(); + + sut.close(); + } + + /** + * Shutdown hooks run concurrently, so the application may keep logging after the flush. Those + * payloads must be sent straight away, since the JVM could exit before the next scheduled run. + */ + @Test + public void shouldSendNewPayloadsRightAwayAfterFlushForShutdown() throws Exception { + Payload beforeFlush = mock(Payload.class); + final Payload afterFlush = mock(Payload.class); + + sut = bufferedSenderWithRealExecutor(new ConcurrentLinkedQueue<>()); + + sut.send(beforeFlush); + verify(sender, after(200).never()).send(beforeFlush); + + sut.flushForShutdown(); + verify(sender).send(beforeFlush); + + sut.send(afterFlush); + verify(sender, timeout(5000)).send(afterFlush); + + sut.close(); + } + + /** + * A retry must still wait for the next scheduled run, otherwise a network outage during + * shutdown would burn through every attempt back to back. + */ + @Test + public void shouldNotRetryRightAwayAfterFlushForShutdown() throws Exception { + final Payload failing = mock(Payload.class); + final Payload fresh = mock(Payload.class); + queue = new ConcurrentLinkedQueue<>(); + + Response response = new Response.Builder().result(new Result.Builder().build()).build(); + setSenderMockToCallListeners(sender, response, null); + + SenderFailureStrategy strategy = mock(SenderFailureStrategy.class); + when(strategy.getAction(any(Payload.class), eq(response))) + .thenReturn(SenderFailureStrategy.PayloadAction.CAN_BE_RETRIED); + + // A fresh mock, so that only this sender's scheduled task gets captured below. + executorService = mock(ScheduledExecutorService.class); + doReturn(CompletableFuture.completedFuture(null)) + .when(executorService).submit(any(Runnable.class)); + + sut = new BufferedSender(new BufferedSender.Builder() + .queue(queue) + .sender(sender) + // One payload per run. The strategy never suspends sending, so a larger batch would keep + // taking the retried payload straight back off the queue. + .batchSize(1) + .senderFailureStrategy(strategy), + executorService); + + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(SendTask.class); + verify(executorService).scheduleWithFixedDelay( + taskCaptor.capture(), anyLong(), anyLong(), any()); + + sut.flushForShutdown(); + + // A scheduled run sends the payload, which fails and is queued again for retry. + queue.add(failing); + taskCaptor.getValue().run(); + + verify(sender).send(failing); + assertThat(queue.peek(), is(failing)); + verify(executorService, never()).execute(any(Runnable.class)); + + sut.send(fresh); + + verify(executorService).execute(any(Runnable.class)); + } + + private BufferedSender bufferedSenderWithRealExecutor(Queue queue) { + return new BufferedSender(new BufferedSender.Builder() + .queue(queue) + .sender(sender) + // Far enough out that no scheduled run gets in the way of the test. + .initialFlushDelay(TimeUnit.MINUTES.toMillis(1)) + .flushFreq(TimeUnit.MINUTES.toMillis(1)), + Executors.newSingleThreadScheduledExecutor(new SenderThreadFactory())); + } + @Test public void threadFactoryShouldSetNameMakeThreadsDaemons() { Runnable runnable = mock(Runnable.class); diff --git a/rollbar-java/src/test/java/com/rollbar/notifier/sender/SenderShutdownHookTest.java b/rollbar-java/src/test/java/com/rollbar/notifier/sender/SenderShutdownHookTest.java new file mode 100644 index 00000000..207c870c --- /dev/null +++ b/rollbar-java/src/test/java/com/rollbar/notifier/sender/SenderShutdownHookTest.java @@ -0,0 +1,210 @@ +package com.rollbar.notifier.sender; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.empty; +import static org.hamcrest.core.Is.is; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; + +import com.rollbar.api.payload.Payload; +import com.rollbar.notifier.provider.Provider; + +import java.lang.Thread.UncaughtExceptionHandler; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.stubbing.Answer; + +public class SenderShutdownHookTest { + + private static final long TIMEOUT = 500L; + + private static final String FLUSH_THREAD_NAME = "rollbar-shutdown-flush"; + + private final List uncaughtOnFlushThread = new CopyOnWriteArrayList<>(); + + private UncaughtExceptionHandler previousHandler; + + /** + * The flush runs on its own thread, so anything it throws can never reach the thread that called + * {@link SenderShutdownHook#run()}. It would end up in the uncaught exception handler instead, + * so that is where these tests look for it. + */ + @Before + public void setUp() { + previousHandler = Thread.getDefaultUncaughtExceptionHandler(); + Thread.setDefaultUncaughtExceptionHandler((thread, error) -> { + if (FLUSH_THREAD_NAME.equals(thread.getName())) { + uncaughtOnFlushThread.add(error); + } else if (previousHandler != null) { + previousHandler.uncaughtException(thread, error); + } + }); + } + + @After + public void tearDown() { + Thread.setDefaultUncaughtExceptionHandler(previousHandler); + } + + @Test + public void shouldFlushTheSender() throws Exception { + Sender sender = mock(Sender.class); + + new SenderShutdownHook(providerOf(sender), TIMEOUT).run(); + + verify(sender).close(true); + } + + @Test + public void shouldReturnWithoutWaitingWhenTimeoutIsNotPositive() throws Exception { + Sender sender = mock(Sender.class); + + long start = System.nanoTime(); + new SenderShutdownHook(providerOf(sender), 0L).run(); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + + assertTrue("Expected not to wait, but waited " + elapsedMillis + " ms", elapsedMillis < 200L); + } + + /** + * The sender performs blocking network I/O without a socket timeout of its own, so the hook has + * to give up on its own rather than let a stalled flush hold JVM shutdown open indefinitely. + */ + @Test + public void shouldGiveUpOnceTheTimeoutElapses() throws Exception { + final CountDownLatch released = new CountDownLatch(1); + Sender sender = mock(Sender.class, (Answer) invocation -> { + if ("close".equals(invocation.getMethod().getName())) { + // Block until the test releases us, simulating an unresponsive endpoint. + released.await(10, TimeUnit.SECONDS); + } + return null; + }); + + long start = System.nanoTime(); + new SenderShutdownHook(providerOf(sender), TIMEOUT).run(); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + + released.countDown(); + + assertTrue("Expected to give up at ~" + TIMEOUT + " ms, but took " + elapsedMillis + " ms", + elapsedMillis >= TIMEOUT && elapsedMillis < TIMEOUT * 6); + } + + /** + * A failure inside the notifier must not surface as a confusing error while the JVM is exiting. + */ + @Test + public void shouldNotPropagateFlushFailures() throws Exception { + Sender sender = mock(Sender.class); + doThrow(new IllegalStateException("network down")).when(sender).close(true); + + new SenderShutdownHook(providerOf(sender), TIMEOUT).run(); + + verify(sender).close(true); + assertThat(uncaughtOnFlushThread, is(empty())); + } + + @Test + public void shouldFlushOnADaemonThreadSoItCanNeverKeepTheJvmAlive() throws Exception { + final AtomicBoolean daemon = new AtomicBoolean(false); + Sender sender = mock(Sender.class, (Answer) invocation -> { + if ("close".equals(invocation.getMethod().getName())) { + daemon.set(Thread.currentThread().isDaemon()); + } + return null; + }); + + new SenderShutdownHook(providerOf(sender), TIMEOUT).run(); + + assertThat(daemon.get(), is(true)); + } + + /** + * The sender is resolved when the hook runs, not when it is registered, so replacing the + * configuration after registration flushes the sender actually in use. + */ + @Test + public void shouldFlushTheSenderInForceWhenTheHookRuns() throws Exception { + Sender original = mock(Sender.class); + final Sender replacement = mock(Sender.class); + final AtomicBoolean replaced = new AtomicBoolean(false); + + Provider provider = () -> replaced.get() ? replacement : original; + Thread hook = new SenderShutdownHook(provider, TIMEOUT); + + replaced.set(true); + hook.run(); + + verify(replacement).close(true); + verify(original, org.mockito.Mockito.never()).close(true); + } + + /** + * Shutdown hooks run concurrently, so the application may still be logging after this hook + * finishes, for example while a web server drains in-flight requests. A buffered sender must + * therefore be flushed without being closed, and keep sending afterwards. + */ + @Test + public void shouldKeepABufferedSenderSendingAfterTheHookRuns() throws Exception { + Sender delegate = mock(Sender.class); + Payload queued = mock(Payload.class); + final Payload loggedLater = mock(Payload.class); + + Queue queue = new ConcurrentLinkedQueue<>(); + queue.add(queued); + BufferedSender sender = new BufferedSender.Builder() + .queue(queue) + .sender(delegate) + // Far enough out that only the hook can be the one sending. + .initialFlushDelay(TimeUnit.MINUTES.toMillis(1)) + .flushFreq(TimeUnit.MINUTES.toMillis(1)) + .build(); + + new SenderShutdownHook(providerOf(sender), TIMEOUT).run(); + + verify(delegate).send(queued); + + sender.send(loggedLater); + + verify(delegate, timeout(5000)).send(loggedLater); + verify(delegate, never()).close(); + assertThat(uncaughtOnFlushThread, is(empty())); + + sender.close(); + } + + /** + * A notifier configured without a sender has nothing to flush, and that must not fail. + */ + @Test + public void shouldDoNothingWhenThereIsNoSender() { + final AtomicBoolean consulted = new AtomicBoolean(false); + Provider provider = () -> { + consulted.set(true); + return null; + }; + + new SenderShutdownHook(provider, TIMEOUT).run(); + + assertThat(consulted.get(), is(true)); + assertThat(uncaughtOnFlushThread, is(empty())); + } + + private Provider providerOf(final Sender sender) { + return () -> sender; + } +}