The watcher indexing listener didn't handle document level exceptions. (#51466)

Prior to the change the watcher index listener didn't implement the
`postIndex(ShardId, Engine.Index, Engine.IndexResult)` method. This
caused document level exceptions like VersionConflictEngineException
to be ignored. This commit fixes this.

The watcher indexing listener did implement the `postIndex(ShardId, Engine.Index, Exception)`
method, but that only handles engine level exceptions.

This change also unmutes the SmokeTestWatcherTestSuiteIT#testMonitorClusterHealth test again.

Relates to #32299

Author: martijnvg

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/WatcherIndexingListener.java

@@ -135,8 +135,19 @@ public Engine.Index preIndex(ShardId shardId, Engine.Index operation) {
     }
 
     /**
-     *
-     * In case of an error, we have to ensure that the triggerservice does not leave anything behind
+     * In case of a document related failure (for example version conflict), then clean up resources for a watch
+     * in the trigger service.
+     */
+    @Override
+    public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
+        if (result.getResultType() == Engine.Result.Type.FAILURE) {
+            assert result.getFailure() != null;
+            postIndex(shardId, index, result.getFailure());
+        }
+    }
+
+    /**
+     * In case of an engine related error, we have to ensure that the triggerservice does not leave anything behind
      *
      * TODO: If the configuration changes between preindex and postindex methods and we add a
      *       watch, that could not be indexed
@@ -153,10 +164,6 @@ public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
         if (isWatchDocument(shardId.getIndexName())) {
             logger.debug(() -> new ParameterizedMessage("removing watch [{}] from trigger", index.id()), ex);
             triggerService.remove(index.id());
-        } else {
-            logger.debug(
-                () -> new ParameterizedMessage("not removing watch [{}] from trigger, because [{}] != [{}] configuration.active=[{}]",
-                index.id(), shardId.getIndexName(), configuration.index, configuration.active), ex);
         }
     }
 

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/WatcherIndexingListenerTests.java

@@ -200,15 +200,48 @@ public void testPreIndexCheckParsingException() throws Exception {
         assertThat(exc.getMessage(), containsString(id));
     }
 
-    public void testPostIndexRemoveTriggerOnException() throws Exception {
+    public void testPostIndexRemoveTriggerOnDocumentRelatedException() throws Exception {
+        when(operation.id()).thenReturn("_id");
+        when(result.getResultType()).thenReturn(Engine.Result.Type.FAILURE);
+        when(result.getFailure()).thenReturn(new RuntimeException());
+        when(shardId.getIndexName()).thenReturn(Watch.INDEX);
+
+        listener.postIndex(shardId, operation, result);
+        verify(triggerService).remove(eq("_id"));
+    }
+
+    public void testPostIndexRemoveTriggerOnDocumentRelatedException_ignoreOtherEngineResultTypes() throws Exception {
+        List<Engine.Result.Type> types = new ArrayList<>(List.of(Engine.Result.Type.values()));
+        types.remove(Engine.Result.Type.FAILURE);
+
+        when(operation.id()).thenReturn("_id");
+        when(result.getResultType()).thenReturn(randomFrom(types));
+        when(result.getFailure()).thenReturn(new RuntimeException());
+        when(shardId.getIndexName()).thenReturn(Watch.INDEX);
+
+        listener.postIndex(shardId, operation, result);
+        verifyZeroInteractions(triggerService);
+    }
+
+    public void testPostIndexRemoveTriggerOnDocumentRelatedException_ignoreNonWatcherDocument() throws Exception {
+        when(operation.id()).thenReturn("_id");
+        when(result.getResultType()).thenReturn(Engine.Result.Type.FAILURE);
+        when(result.getFailure()).thenReturn(new RuntimeException());
+        when(shardId.getIndexName()).thenReturn(randomAlphaOfLength(4));
+
+        listener.postIndex(shardId, operation, result);
+        verifyZeroInteractions(triggerService);
+    }
+
+    public void testPostIndexRemoveTriggerOnEngineLevelException() throws Exception {
         when(operation.id()).thenReturn("_id");
         when(shardId.getIndexName()).thenReturn(Watch.INDEX);
 
         listener.postIndex(shardId, operation, new ElasticsearchParseException("whatever"));
         verify(triggerService).remove(eq("_id"));
     }
 
-    public void testPostIndexDontInvokeForOtherDocuments() throws Exception {
+    public void testPostIndexRemoveTriggerOnEngineLevelException_ignoreNonWatcherDocument() throws Exception {
         when(operation.id()).thenReturn("_id");
         when(shardId.getIndexName()).thenReturn("anything");
         when(result.getResultType()).thenReturn(Engine.Result.Type.SUCCESS);

x-pack/qa/smoke-test-watcher/src/test/java/org/elasticsearch/smoketest/SmokeTestWatcherTestSuiteIT.java

@@ -110,7 +110,6 @@ protected Settings restAdminSettings() {
         return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", token).build();
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/32299")
     public void testMonitorClusterHealth() throws Exception {
         final String watchId = "cluster_health_watch";