From 99d355de538007e14da660e1811257e696e28324 Mon Sep 17 00:00:00 2001 From: Henning Andersen Date: Wed, 29 Jan 2020 20:51:26 +0100 Subject: [PATCH 1/2] [ML] Fix possible race condition starting datafeed Datafeeds being closed while starting could result in and NPE. This was handled as any other failure, masking out the NPE. However, this conflicts with the changes in #50886. Related to #50886 and #51302 --- .../elasticsearch/xpack/ml/datafeed/DatafeedManager.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java index e4b0940803271..aa8cb1dcf278b 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java @@ -520,7 +520,12 @@ private void runTask(TransportStartDatafeedAction.DatafeedTask task) { // a context with sufficient permissions would coincidentally be in force in some single node // tests, leading to bugs not caught in CI due to many tests running in single node test clusters. try (ThreadContext.StoredContext ignore = threadPool.getThreadContext().stashContext()) { - innerRun(runningDatafeedsOnThisNode.get(task.getAllocationId()), task.getDatafeedStartTime(), task.getEndTime()); + Holder holder = runningDatafeedsOnThisNode.get(task.getAllocationId()); + if (holder != null) { + innerRun(holder, task.getDatafeedStartTime(), task.getEndTime()); + } else { + logger.warn("Datafeed [{}] was closed while being opened", task.getDatafeedId()); + } } } From b8978dd8cb9e8cdae19f2d079caee1b335547c46 Mon Sep 17 00:00:00 2001 From: Henning Andersen Date: Wed, 29 Jan 2020 21:06:46 +0100 Subject: [PATCH 2/2] Fix message according to review. --- .../org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java index aa8cb1dcf278b..b486916be690c 100644 --- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java +++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedManager.java @@ -524,7 +524,7 @@ private void runTask(TransportStartDatafeedAction.DatafeedTask task) { if (holder != null) { innerRun(holder, task.getDatafeedStartTime(), task.getEndTime()); } else { - logger.warn("Datafeed [{}] was closed while being opened", task.getDatafeedId()); + logger.warn("Datafeed [{}] was stopped while being started", task.getDatafeedId()); } } }