|
@@ -12,6 +12,7 @@ import org.apache.lucene.util.SetOnce;
|
|
|
import org.elasticsearch.action.admin.indices.refresh.RefreshAction;
|
|
|
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
|
|
|
import org.elasticsearch.action.search.SearchResponse;
|
|
|
+import org.elasticsearch.action.support.IndicesOptions;
|
|
|
import org.elasticsearch.client.Client;
|
|
|
import org.elasticsearch.common.Nullable;
|
|
|
import org.elasticsearch.common.Strings;
|
|
@@ -159,6 +160,7 @@ public class AnalyticsProcessManager {
|
|
|
processContext.setFailureReason(resultProcessor.getFailure());
|
|
|
|
|
|
refreshDest(config);
|
|
|
+ refreshStateIndex(config.getId());
|
|
|
LOGGER.info("[{}] Result processor has completed", config.getId());
|
|
|
} catch (Exception e) {
|
|
|
if (task.isStopping()) {
|
|
@@ -288,6 +290,17 @@ public class AnalyticsProcessManager {
|
|
|
() -> client.execute(RefreshAction.INSTANCE, new RefreshRequest(config.getDest().getIndex())).actionGet());
|
|
|
}
|
|
|
|
|
|
+ private void refreshStateIndex(String jobId) {
|
|
|
+ String indexName = AnomalyDetectorsIndex.jobStateIndexPattern();
|
|
|
+ LOGGER.debug("[{}] Refresh index {}", jobId, indexName);
|
|
|
+
|
|
|
+ RefreshRequest refreshRequest = new RefreshRequest(indexName);
|
|
|
+ refreshRequest.indicesOptions(IndicesOptions.lenientExpandOpen());
|
|
|
+ try (ThreadContext.StoredContext ignore = client.threadPool().getThreadContext().stashWithOrigin(ML_ORIGIN)) {
|
|
|
+ client.admin().indices().refresh(refreshRequest).actionGet();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
private void closeProcess(DataFrameAnalyticsTask task) {
|
|
|
String configId = task.getParams().getId();
|
|
|
LOGGER.info("[{}] Closing process", configId);
|