From 2f142409155f56948f70112047c5756465bd975d Mon Sep 17 00:00:00 2001
From: Martijn van Groningen <martijn.v.groningen@gmail.com>
Date: Wed, 17 Jun 2015 09:17:13 +0200
Subject: [PATCH 1/2] test: increased time throttle timeout to deal with timing
 issues

Original commit: elastic/x-pack-elasticsearch@90fd3c9cd956321c899c3f8530f43dfa458cac99
---
 .../actions/throttler/ActionThrottleTests.java      | 13 ++++++-------
 1 file changed, 6 insertions(+), 7 deletions(-)

diff --git a/src/test/java/org/elasticsearch/watcher/actions/throttler/ActionThrottleTests.java b/src/test/java/org/elasticsearch/watcher/actions/throttler/ActionThrottleTests.java
index 8cdd31f42c1..d2930396fed 100644
--- a/src/test/java/org/elasticsearch/watcher/actions/throttler/ActionThrottleTests.java
+++ b/src/test/java/org/elasticsearch/watcher/actions/throttler/ActionThrottleTests.java
@@ -5,7 +5,6 @@
  */
 package org.elasticsearch.watcher.actions.throttler;
 
-import com.carrotsearch.randomizedtesting.annotations.Repeat;
 import org.apache.lucene.util.LuceneTestCase.Slow;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.joda.time.DateTime;
@@ -56,7 +55,7 @@ import static org.hamcrest.Matchers.greaterThan;
  */
 public class ActionThrottleTests extends AbstractWatcherIntegrationTests {
 
-    @Test @Slow @Repeat(iterations = 10)
+    @Test @Slow
     public void testSingleActionAckThrottle() throws Exception {
         boolean useClientForAcking = randomBoolean();
 
@@ -96,7 +95,7 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTests {
         }
     }
 
-    @Test @Slow @Repeat(iterations = 10)
+    @Test @Slow
     public void testRandomMultiActionAckThrottle() throws Exception {
         boolean useClientForAcking = randomBoolean();
 
@@ -256,11 +255,11 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTests {
         }, 6, TimeUnit.SECONDS);
     }
 
-    @Test @Slow @Repeat(iterations = 20)
+    @Test @Slow
     public void testWatchThrottlePeriod() throws Exception {
         WatchSourceBuilder watchSourceBuilder = watchBuilder()
                 .trigger(schedule(interval("60m")))
-                .defaultThrottlePeriod(new TimeValue(1, TimeUnit.SECONDS));
+                .defaultThrottlePeriod(new TimeValue(20, TimeUnit.SECONDS));
 
         AvailableAction availableAction = randomFrom(AvailableAction.values());
         watchSourceBuilder.addAction("default_global_throttle", availableAction.action());
@@ -296,7 +295,7 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTests {
         assertThat(resultStatus.toString(), equalTo("throttled"));
 
         if (timeWarped()) {
-            timeWarp().clock().fastForwardSeconds(1);
+            timeWarp().clock().fastForwardSeconds(20);
         }
         assertBusy(new Runnable() {
             @Override
@@ -315,7 +314,7 @@ public class ActionThrottleTests extends AbstractWatcherIntegrationTests {
                     throw new ElasticsearchException("failed to execute", ioe);
                 }
             }
-        }, 1, TimeUnit.SECONDS);
+        }, 20, TimeUnit.SECONDS);
     }
 
     @Test @Slow

From b82fca337912d150e4da4f05f4bcca169b59aa31 Mon Sep 17 00:00:00 2001
From: Martijn van Groningen <martijn.v.groningen@gmail.com>
Date: Wed, 17 Jun 2015 09:47:30 +0200
Subject: [PATCH 2/2] We should cope with the situation when a watch is going
 to be executed whilst shutting down.

The execution service will throw an exection if we acquire a watch lock or add a watch to the current executions if Watcher has stopped or is stopping. We should deal with this sitation by properly catching those failures. FYI if a watch managed to put itself to the current executions a shutdown will wait until a watch execution is finished.

Original commit: elastic/x-pack-elasticsearch@5692316072c73c0b61e0c055f34f4e35c9bdeb31
---
 .../watcher/execution/CurrentExecutions.java  |  5 ++---
 .../watcher/execution/ExecutionService.java   | 18 ++++++++++--------
 .../support/WatcherInactiveException.java     | 19 +++++++++++++++++++
 .../watcher/watch/WatchLockService.java       |  6 +++---
 4 files changed, 34 insertions(+), 14 deletions(-)
 create mode 100644 src/main/java/org/elasticsearch/watcher/support/WatcherInactiveException.java

diff --git a/src/main/java/org/elasticsearch/watcher/execution/CurrentExecutions.java b/src/main/java/org/elasticsearch/watcher/execution/CurrentExecutions.java
index 959e436dd31..a3619ba0fbb 100644
--- a/src/main/java/org/elasticsearch/watcher/execution/CurrentExecutions.java
+++ b/src/main/java/org/elasticsearch/watcher/execution/CurrentExecutions.java
@@ -5,9 +5,8 @@
  */
 package org.elasticsearch.watcher.execution;
 
-import org.elasticsearch.ElasticsearchIllegalStateException;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
+import org.elasticsearch.watcher.support.WatcherInactiveException;
 
 import java.util.Iterator;
 import java.util.concurrent.ConcurrentHashMap;
@@ -28,7 +27,7 @@ public class CurrentExecutions implements Iterable<ExecutionService.WatchExecuti
         try {
             if (seal) {
                 // We shouldn't get here, because, ExecutionService#started should have been set to false
-                throw new ElasticsearchIllegalStateException("put execution forbidden, because we're sealed");
+                throw new WatcherInactiveException("could not register execution [{}]. current executions are sealed and forbid registrations of additional executions.", id);
             }
             currentExecutions.put(id, execution);
         } finally {
diff --git a/src/main/java/org/elasticsearch/watcher/execution/ExecutionService.java b/src/main/java/org/elasticsearch/watcher/execution/ExecutionService.java
index a0a690ae4e0..e9774408f50 100644
--- a/src/main/java/org/elasticsearch/watcher/execution/ExecutionService.java
+++ b/src/main/java/org/elasticsearch/watcher/execution/ExecutionService.java
@@ -21,6 +21,7 @@ import org.elasticsearch.watcher.condition.Condition;
 import org.elasticsearch.watcher.history.HistoryStore;
 import org.elasticsearch.watcher.history.WatchRecord;
 import org.elasticsearch.watcher.input.Input;
+import org.elasticsearch.watcher.support.WatcherInactiveException;
 import org.elasticsearch.watcher.support.clock.Clock;
 import org.elasticsearch.watcher.support.validation.WatcherSettingsValidation;
 import org.elasticsearch.watcher.trigger.TriggerEvent;
@@ -237,9 +238,7 @@ public class ExecutionService extends AbstractComponent {
             logger.trace("acquired lock for [{}] -- [{}]", ctx.id(), System.identityHashCode(lock));
         }
         try {
-
             currentExecutions.put(ctx.watch().id(), new WatchExecution(ctx, Thread.currentThread()));
-
             if (ctx.knownWatch() && watchStore.get(ctx.watch().id()) == null) {
                 // fail fast if we are trying to execute a deleted watch
                 String message = "unable to find watch for record [" + ctx.id() + "], perhaps it has been deleted, ignoring...";
@@ -252,14 +251,12 @@ public class ExecutionService extends AbstractComponent {
                     watchStore.updateStatus(ctx.watch());
                 }
             }
-
         } catch (Exception e) {
             String detailedMessage = ExceptionsHelper.detailedMessage(e);
             logger.warn("failed to execute watch [{}], failure [{}]", ctx.id(), detailedMessage);
             record = ctx.abortFailedExecution(detailedMessage);
 
         } finally {
-
             if (ctx.knownWatch() && record != null && ctx.recordExecution()) {
                 try {
                     historyStore.put(record);
@@ -267,13 +264,11 @@ public class ExecutionService extends AbstractComponent {
                     logger.error("failed to update watch record [{}]", e, ctx.id());
                 }
             }
-
             try {
                 triggeredWatchStore.delete(ctx.id());
             } catch (Exception e) {
                 logger.error("failed to delete triggered watch [{}]", e, ctx.id());
             }
-
             currentExecutions.remove(ctx.watch().id());
             if (logger.isTraceEnabled()) {
                 logger.trace("releasing lock for [{}] -- [{}]", ctx.id(), System.identityHashCode(lock));
@@ -281,7 +276,6 @@ public class ExecutionService extends AbstractComponent {
             lock.release();
             logger.trace("finished [{}]/[{}]", ctx.watch().id(), ctx.id());
         }
-
         return record;
     }
 
@@ -375,7 +369,15 @@ public class ExecutionService extends AbstractComponent {
 
         @Override
         public void run() {
-            execute(ctx);
+            try {
+                execute(ctx);
+            } catch (WatcherInactiveException e) {
+                // When can end up here when acquiring the lock or adding a watch to the current executions while shutting down.
+                // Once we a watch is added to the current executions we shouldn't end up here.
+                logger.debug("could not execute watch [{}]/[{}]. watcher is not active", e, ctx.watch().id(), ctx.id());
+            } catch (Exception e) {
+                logger.error("could not execute watch [{}]/[{}]", e, ctx.watch().id(), ctx.id());
+            }
         }
     }
 
diff --git a/src/main/java/org/elasticsearch/watcher/support/WatcherInactiveException.java b/src/main/java/org/elasticsearch/watcher/support/WatcherInactiveException.java
new file mode 100644
index 00000000000..df30694db19
--- /dev/null
+++ b/src/main/java/org/elasticsearch/watcher/support/WatcherInactiveException.java
@@ -0,0 +1,19 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.watcher.support;
+
+import org.elasticsearch.watcher.WatcherException;
+
+public class WatcherInactiveException extends WatcherException {
+
+    public WatcherInactiveException(String msg, Object... args) {
+        super(msg, args);
+    }
+
+    public WatcherInactiveException(String msg, Throwable cause, Object... args) {
+        super(msg, cause, args);
+    }
+}
diff --git a/src/main/java/org/elasticsearch/watcher/watch/WatchLockService.java b/src/main/java/org/elasticsearch/watcher/watch/WatchLockService.java
index f71a909a6d1..c255602f48c 100644
--- a/src/main/java/org/elasticsearch/watcher/watch/WatchLockService.java
+++ b/src/main/java/org/elasticsearch/watcher/watch/WatchLockService.java
@@ -5,7 +5,6 @@
  */
 package org.elasticsearch.watcher.watch;
 
-import org.elasticsearch.ElasticsearchIllegalStateException;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.joda.time.PeriodType;
@@ -13,6 +12,7 @@ import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.watcher.WatcherException;
+import org.elasticsearch.watcher.support.WatcherInactiveException;
 import org.elasticsearch.watcher.support.concurrent.FairKeyedLock;
 
 import java.util.concurrent.TimeUnit;
@@ -43,7 +43,7 @@ public class WatchLockService extends AbstractComponent {
 
     public Lock acquire(String name) {
         if (!running.get()) {
-            throw new ElasticsearchIllegalStateException("not started");
+            throw new WatcherInactiveException("not started");
         }
 
         watchLocks.acquire(name);
@@ -52,7 +52,7 @@ public class WatchLockService extends AbstractComponent {
 
     public Lock tryAcquire(String name, TimeValue timeout) {
         if (!running.get()) {
-            throw new ElasticsearchIllegalStateException("not started");
+            throw new WatcherInactiveException("not started");
         }
         try {
             if (!watchLocks.tryAcquire(name, timeout.millis(), TimeUnit.MILLISECONDS)) {