MlAssignmentNotifier cannot read config from state

Author: davidkyle

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlAssignmentNotifier.java

@@ -12,15 +12,13 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
+import org.elasticsearch.persistent.PersistentTasksCustomMetaData.Assignment;
+import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.xpack.core.ml.MlMetadata;
 import org.elasticsearch.xpack.core.ml.MlTasks;
 import org.elasticsearch.xpack.core.ml.action.OpenJobAction;
 import org.elasticsearch.xpack.core.ml.action.StartDatafeedAction;
-import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
-import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
-import org.elasticsearch.persistent.PersistentTasksCustomMetaData.Assignment;
-import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
 import org.elasticsearch.xpack.ml.notifications.Auditor;
 
 import java.util.Objects;
@@ -89,16 +87,20 @@ public void clusterChanged(ClusterChangedEvent event) {
                     auditor.info(jobId, "Opening job on node [" + node.toString() + "]");
                 }
             } else if (MlTasks.DATAFEED_TASK_NAME.equals(currentTask.getTaskName())) {
-                String datafeedId = ((StartDatafeedAction.DatafeedParams) currentTask.getParams()).getDatafeedId();
-                DatafeedConfig datafeedConfig = MlMetadata.getMlMetadata(event.state()).getDatafeed(datafeedId);
+                StartDatafeedAction.DatafeedParams datafeedParams = (StartDatafeedAction.DatafeedParams) currentTask.getParams();
+                String jobId = datafeedParams.getJob() != null ? datafeedParams.getJob().getId() : null;
                 if (currentAssignment.getExecutorNode() == null) {
-                    String msg = "No node found to start datafeed [" + datafeedId +"]. Reasons [" +
+                    String msg = "No node found to start datafeed [" + datafeedParams.getDatafeedId() +"]. Reasons [" +
                             currentAssignment.getExplanation() + "]";
-                    logger.warn("[{}] {}", datafeedConfig.getJobId(), msg);
-                    auditor.warning(datafeedConfig.getJobId(), msg);
+                    logger.warn("[{}] {}", jobId, msg);
+                    if (jobId != null) {
+                        auditor.warning(jobId, msg);
+                    }
                 } else {
                     DiscoveryNode node = event.state().nodes().get(currentAssignment.getExecutorNode());
-                    auditor.info(datafeedConfig.getJobId(), "Starting datafeed [" + datafeedId + "] on node [" + node + "]");
+                    if (jobId != null) {
+                        auditor.info(jobId, "Starting datafeed [" + datafeedParams.getDatafeedId() + "] on node [" + node + "]");
+                    }
                 }
             }
         }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlAssignmentNotifierTests.java

@@ -32,7 +32,7 @@
 
 public class MlAssignmentNotifierTests extends ESTestCase {
 
-    public void testClusterChanged_info() throws Exception {
+    public void testClusterChanged_info() {
         Auditor auditor = mock(Auditor.class);
         ClusterService clusterService = mock(ClusterService.class);
         MlAssignmentNotifier notifier = new MlAssignmentNotifier(Settings.EMPTY, auditor, clusterService);
@@ -60,7 +60,7 @@ public void testClusterChanged_info() throws Exception {
         verifyNoMoreInteractions(auditor);
     }
 
-    public void testClusterChanged_warning() throws Exception {
+    public void testClusterChanged_warning() {
         Auditor auditor = mock(Auditor.class);
         ClusterService clusterService = mock(ClusterService.class);
         MlAssignmentNotifier notifier = new MlAssignmentNotifier(Settings.EMPTY, auditor, clusterService);