Wait for confirms of published events and add them to cache

Solves: Jira GER-1561
Change-Id: Ie0a41d2db66bfd49f01e0721c1774759d0f7f8a6

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

diff --git a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java
index 8db4ce8..96e4093 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/EiffelEventHubImpl.java
@@ -205,22 +205,36 @@
     final ReentrantLock takeLock = this.takeLock;
     final ReentrantLock idLookupLock = this.idLookupLock;
     int previousCount = -1;
-    takeLock.lock();
-    idLookupLock.lock();
     try {
-      taken = null;
-      previousCount = count.get();
-      for (EiffelEvent event : events) {
-        idCache.putId(event);
-        EventKey key = EventKey.fromEvent(event);
-        eventsInQueue.remove(key);
-        count.decrementAndGet();
-        logger.atFine().log("Event %s have been ack:ed by publisher", key);
+      takeLock.lockInterruptibly();
+      idLookupLock.lockInterruptibly();
+      try {
+        taken = null;
+        previousCount = count.get();
+        for (EiffelEvent event : events) {
+          idCache.putId(event);
+          EventKey key = EventKey.fromEvent(event);
+          eventsInQueue.remove(key);
+          count.decrementAndGet();
+          logger.atFine().log("Event %s have been ack:ed by publisher", key);
+        }
+        readyForTake.signal();
+      } finally {
+        takeLock.unlock();
+        idLookupLock.unlock();
       }
-      readyForTake.signal();
-    } finally {
-      takeLock.unlock();
-      idLookupLock.unlock();
+    } catch (InterruptedException ie) {
+      logger.atWarning().log("Interupted while ack:ing events, attempting to clean up.");
+      for (EiffelEvent event : events) {
+        EventKey key = EventKey.fromEvent(event);
+        if (eventsInQueue.containsKey(key)) {
+          idCache.putId(event);
+          eventsInQueue.remove(key);
+          count.decrementAndGet();
+          logger.atFine().log("Event %s have been ack:ed by publisher", key);
+        }
+      }
+      logger.atInfo().log("Cleanup complete.");
     }
     if (previousCount == MAX_SIZE) {
       signalReadyForPut();

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 b192bb3..34f4e5d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/PublishEventWorker.java
@@ -41,10 +41,10 @@
     try {
       while (running) {
         events = eventQueue.take(publisher.maxBatchSize());
-        boolean unpublishedEvent = !events.isEmpty();
-        while (running && unpublishedEvent) {
-          unpublishedEvent = !publisher.publish(events);
-          if (!unpublishedEvent) {
+        boolean unpublishedEvents = !events.isEmpty();
+        while (running && unpublishedEvents) {
+          unpublishedEvents = !publisher.publish(events);
+          if (!unpublishedEvents) {
             eventQueue.ack(events);
             events = null;
           } else {
@@ -83,7 +83,11 @@
   public void stop() {
     logger.atInfo().log("Stopping publisher thread");
     this.running = false;
-    thread.interrupt();
+    try {
+      thread.join(5000);
+    } catch (InterruptedException e) {
+      // Do nothing
+    }
   }
 
   @Override

src/main/java/com/googlesource/gerrit/plugins/eventseiffel/mq/RabbitMqPublisher.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/mq/RabbitMqPublisher.java b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/mq/RabbitMqPublisher.java
index 564c91e..7d45b66 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/mq/RabbitMqPublisher.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/mq/RabbitMqPublisher.java
@@ -26,6 +26,7 @@
 import com.rabbitmq.client.Connection;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.concurrent.TimeoutException;
 
@@ -63,9 +64,20 @@
       if (config.confirmsEnabled()) {
         try {
           return channel.waitForConfirms(config.waitForConfirms());
-        } catch (InterruptedException | TimeoutException e) {
-          logger.atWarning().withCause(e).log("Failed to confirm delivery");
+        } catch (TimeoutException te) {
+          logger.atWarning().withCause(te).log("Failed to confirm delivery");
           return false;
+        } catch (InterruptedException ie) {
+          logger.atWarning().log("Interupted while waiting for confirms, new attempt");
+          try {
+            channel.waitForConfirms(2000);
+            return true;
+          } catch (InterruptedException | TimeoutException e) {
+            logger.atSevere().withCause(e).log(
+                "Failed to confirm pushed events before cleanup: %s",
+                Arrays.toString(events.toArray()));
+            return false;
+          }
         }
       }
       return true;