Only flush Watcher's bulk processor if Watcher is enabled (#38803)

When shutting down Watcher, the `bulkProcessor` is null if watcher has been
disabled in the configuration. This protects the flush and close calls with a
check for watcher enabled to avoid a NullPointerException

Resolves #38798

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

@@ -697,10 +697,12 @@ public List<ScriptContext<?>> getContexts() {
 
     @Override
     public void close() throws IOException {
-        bulkProcessor.flush();
+        if (enabled) {
+            bulkProcessor.flush();
+        }
         IOUtils.closeWhileHandlingException(httpClient);
         try {
-            if (bulkProcessor.awaitClose(10, TimeUnit.SECONDS) == false) {
+            if (enabled && bulkProcessor.awaitClose(10, TimeUnit.SECONDS) == false) {
                 logger.warn("failed to properly close watcher bulk processor");
             }
         } catch (InterruptedException e) {

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

@@ -17,6 +17,7 @@
 import org.elasticsearch.xpack.core.watcher.watch.Watch;
 import org.elasticsearch.xpack.watcher.notification.NotificationService;
 
+import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
 
@@ -79,7 +80,7 @@ public void testThreadPoolSize() {
         assertThat(Watcher.getWatcherThreadPoolSize(noDataNodeSettings), is(1));
     }
 
-    public void testWatcherDisabledTests() {
+    public void testWatcherDisabledTests() throws IOException {
         Settings settings = Settings.builder()
                 .put("xpack.watcher.enabled", false)
                 .put("path.home", createTempDir())
@@ -102,6 +103,8 @@ public void testWatcherDisabledTests() {
 
         // also no component creation if not enabled
         assertThat(watcher.createComponents(null, null, null, null, null, null, null, null, null), hasSize(0));
+
+        watcher.close();
     }
 
     public void testReload() {