Retry to publish events indefinitely

When the plugin receives a ShutdownSignalException due to flaky LDAP
connections returning 403 Forbidden, it should not stop attempting to
publish events. Instead enter a 30s retry loop.

This will improve resiliency to network shakiness.

Solves: Jira GER-2002
Change-Id: Ia723c30099f17568a534f11b120088b14fe00a8d

src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
index 18e7434..418e2b4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
@@ -14,11 +14,17 @@
 
 package com.googlesource.gerrit.plugins.eventseiffel;
 
+import com.github.rholder.retry.RetryException;
+import com.github.rholder.retry.Retryer;
+import com.github.rholder.retry.RetryerBuilder;
+import com.github.rholder.retry.WaitStrategies;
 import com.google.common.flogger.FluentLogger;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.eventseiffel.eiffel.api.EiffelEventPublisher;
 import com.googlesource.gerrit.plugins.eventseiffel.eiffel.dto.EiffelEvent;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 
 public class PublishEventWorker implements Runnable, EiffelEventHub.Consumer {
   static final String THREAD_NAME = "eiffel-event-publishing";
@@ -43,7 +49,7 @@
         events = eventQueue.take(publisher.maxBatchSize());
         boolean unpublishedEvents = events != null && !events.isEmpty();
         while (running && unpublishedEvents) {
-          unpublishedEvents = !publisher.publish(events);
+          unpublishedEvents = !publishWithRetry(events);
           if (!unpublishedEvents) {
             eventQueue.ack(events);
             events = null;
@@ -53,6 +59,9 @@
           }
         }
       }
+    } catch (RetryException | ExecutionException e) {
+      logger.atSevere().withCause(e).log("Publishing failed even after retries.");
+      running = false;
     } catch (InterruptedException e) {
       if (running) {
         logger.atSevere().withCause(e).log("Publisher thread interrupted while running.");
@@ -67,6 +76,29 @@
     }
   }
 
+  private boolean publishWithRetry(Set<EiffelEvent> events)
+      throws RetryException, ExecutionException {
+    int timeout = 30;
+    Retryer<Boolean> retryer =
+        RetryerBuilder.<Boolean>newBuilder()
+            .retryIfException()
+            .withWaitStrategy(WaitStrategies.fixedWait(timeout, TimeUnit.SECONDS))
+            .build();
+    return retryer.call(
+        () -> {
+          try {
+            return publisher.publish(events);
+          } catch (Exception e) {
+            logger.atWarning().withCause(e).log(
+                "Publishing event failed. Will retry again after %d seconds", timeout);
+            if (running) {
+              throw e;
+            }
+            return false;
+          }
+        });
+  }
+
   @Override
   public void start(EiffelEventQueue queue) {
     if (thread != null && thread.isAlive()) {