[ML] Uses AbstractRunnable for running datafeed, so that we handle rejected execution exceptions.
Original commit: elastic/x-pack-elasticsearch@d04d3fa40166b95d8833ea8a31cbfbfeeb0dde89
@@ -14,6 +14,7 @@ import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.common.util.concurrent.FutureUtils;
import org.elasticsearch.index.mapper.DateFieldMapper;
import org.elasticsearch.rest.RestStatus;
@@ -128,7 +129,16 @@ public class DatafeedJobRunner extends AbstractComponent {
logger.info("Starting datafeed [{}] for job [{}] in [{}, {})", holder.datafeed.getId(), holder.datafeed.getJobId(),
DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.printer().print(startTime),
endTime == null ? INF_SYMBOL : DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.printer().print(endTime));
- holder.future = threadPool.executor(MachineLearning.DATAFEED_RUNNER_THREAD_POOL_NAME).submit(() -> {
+ holder.future = threadPool.executor(MachineLearning.DATAFEED_RUNNER_THREAD_POOL_NAME).submit(new AbstractRunnable() {
+
+ @Override
+ public void onFailure(Exception e) {
+ logger.error("Failed lookback import for job [" + holder.datafeed.getJobId() + "]", e);
+ holder.stop("general_lookback_failure", TimeValue.timeValueSeconds(20), e);
+ }
+
+ @Override
+ protected void doRun() throws Exception {
Long next = null;
try {
next = holder.executeLoopBack(startTime, endTime);
@@ -163,6 +173,7 @@ public class DatafeedJobRunner extends AbstractComponent {
holder.stop("no_realtime", TimeValue.timeValueSeconds(20), null);
holder.problemTracker.finishReport();
}
+ }
});
}
@@ -170,7 +181,16 @@ public class DatafeedJobRunner extends AbstractComponent {
if (holder.isRunning()) {
TimeValue delay = computeNextDelay(delayInMsSinceEpoch);
logger.debug("Waiting [{}] before executing next realtime import for job [{}]", delay, jobId);
- holder.future = threadPool.schedule(delay, MachineLearning.DATAFEED_RUNNER_THREAD_POOL_NAME, () -> {
+ holder.future = threadPool.schedule(delay, MachineLearning.DATAFEED_RUNNER_THREAD_POOL_NAME, new AbstractRunnable() {
+
+ @Override
+ public void onFailure(Exception e) {
+ logger.error("Unexpected datafeed failure for job [" + jobId + "] stopping...", e);
+ holder.stop("general_realtime_error", TimeValue.timeValueSeconds(20), e);
+ }
+
+ @Override
+ protected void doRun() throws Exception {
long nextDelayInMsSinceEpoch;
try {
nextDelayInMsSinceEpoch = holder.executeRealTime();
@@ -193,6 +213,7 @@ public class DatafeedJobRunner extends AbstractComponent {
if (nextDelayInMsSinceEpoch >= 0) {
doDatafeedRealtime(nextDelayInMsSinceEpoch, jobId, holder);
}
+ }
});
}
}
