add exponential backoff for statefultailer when the file doesnt exist (#90)

Author: BrandonArp

src/main/java/com/arpnetworking/metrics/common/tailer/StatefulTailer.java

@@ -110,9 +110,11 @@ protected boolean isRunning() {
 
     private void fileLoop() {
         InitialPosition nextInitialPosition = _initialPosition;
+        int openFileAttempt = 0;
         try {
             while (isRunning()) {
                 // Attempt to open the file
+                openFileAttempt++;
                 try (SeekableByteChannel reader = Files.newByteChannel(_file, StandardOpenOption.READ)) {
                     LOGGER.trace()
                             .setMessage("Opened file")
@@ -131,9 +133,10 @@ private void fileLoop() {
 
                     // Reset per file state
                     _hash = Optional.empty();
+                    openFileAttempt = 0;
                 } catch (final NoSuchFileException e) {
                     _listener.fileNotFound();
-                    _trigger.waitOnTrigger();
+                    _trigger.waitOnFileNotFoundTrigger(openFileAttempt);
                 }
             }
         // Clients may elect to kill the stateful tailer on an exception by calling stop, or they
@@ -312,7 +315,7 @@ private void readLoop(final SeekableByteChannel reader) throws IOException, Inte
                     // rotated.
 
                     // Read interval
-                    _trigger.waitOnTrigger();
+                    _trigger.waitOnReadTrigger();
                 }
             }
 
@@ -347,7 +350,7 @@ private Attributes getAttributes(final Path file, final Optional<Long> lastCheck
     private void rotate(final Optional<SeekableByteChannel> reader, final String reason) throws InterruptedException, IOException {
         // Allow a full read interval before calling it quits on the old file
         if (reader.isPresent()) {
-            _trigger.waitOnTrigger();
+            _trigger.waitOnReadTrigger();
             readLines(reader.get());
         }
 

src/main/java/com/arpnetworking/utility/TimerTrigger.java

@@ -36,10 +36,19 @@ public TimerTrigger(final Duration duration) {
     }
 
     @Override
-    public void waitOnTrigger() throws InterruptedException {
+    public void waitOnReadTrigger() throws InterruptedException {
         Thread.sleep(_duration.getMillis());
     }
 
+    @Override
+    public void waitOnFileNotFoundTrigger(final int attempt) throws InterruptedException {
+        // Max time = 1.3^n * base  (n capped at 20)
+        final double maxBackoff = Math.pow(1.3, Math.min(attempt, 20)) * _duration.getMillis();
+        // Sleep duration is random from 0 to max, capped at 30 seconds
+        final int sleepDurationMillis = (int) Math.max(Math.random() * maxBackoff, 30000);
+        Thread.sleep(sleepDurationMillis);
+    }
+
     /**
      * Generate a Steno log compatible representation.
      *

src/main/java/com/arpnetworking/utility/Trigger.java

@@ -26,5 +26,13 @@ public interface Trigger {
      *
      * @throws InterruptedException thrown when the wait is interrupted.
      */
-    void waitOnTrigger() throws InterruptedException;
+    void waitOnReadTrigger() throws InterruptedException;
+
+    /**
+     * Blocks the current thread.
+     *
+     * @param attempt The attempt number to open or find the file. Used for exponential backoff.
+     * @throws InterruptedException thrown when the wait is interrupted.
+     */
+    void waitOnFileNotFoundTrigger(int attempt) throws InterruptedException;
 }

src/test/java/com/arpnetworking/utility/ManualSingleThreadedTrigger.java

@@ -31,7 +31,15 @@
  */
 public class ManualSingleThreadedTrigger implements Trigger {
     @Override
-    public void waitOnTrigger() throws InterruptedException {
+    public void waitOnReadTrigger() throws InterruptedException {
+        if (_enabled) {
+            _waiter.release();
+            _semaphore.acquire();
+        }
+    }
+
+    @Override
+    public void waitOnFileNotFoundTrigger(final int attempt) throws InterruptedException {
         if (_enabled) {
             _waiter.release();
             _semaphore.acquire();