Skip to content

Commit 0db0e13

Browse files
authored
[ML Data Frame] Set DF task state when stopping (#42516)
Set the state to stopped prior to persisting
1 parent 4a08b3d commit 0db0e13

File tree

3 files changed

+10
-26
lines changed

3 files changed

+10
-26
lines changed

x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/DataFrameTransformPersistentTasksExecutor.java

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@
3030
import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformConfig;
3131
import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformState;
3232
import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformStateAndStats;
33-
import org.elasticsearch.xpack.core.dataframe.transforms.DataFrameTransformTaskState;
3433
import org.elasticsearch.xpack.core.indexing.IndexerState;
3534
import org.elasticsearch.xpack.core.scheduler.SchedulerEngine;
3635
import org.elasticsearch.xpack.dataframe.DataFrame;
@@ -223,18 +222,8 @@ private void startTask(DataFrameTransformTask buildTask,
223222
DataFrameTransformTask.ClientDataFrameIndexerBuilder indexerBuilder,
224223
Long previousCheckpoint,
225224
ActionListener<StartDataFrameTransformTaskAction.Response> listener) {
226-
// If we are stopped, and it is an initial run, this means we have never been started,
227-
// attempt to start the task
228-
229225
buildTask.initializeIndexer(indexerBuilder);
230-
// TODO isInitialRun is false after relocation??
231-
if (buildTask.getState().getTaskState().equals(DataFrameTransformTaskState.STOPPED) && buildTask.isInitialRun()) {
232-
logger.info("Data frame transform [{}] created.", buildTask.getTransformId());
233-
buildTask.start(previousCheckpoint, listener);
234-
} else {
235-
logger.debug("No need to start task. Its current state is: {}", buildTask.getState().getIndexerState());
236-
listener.onResponse(new StartDataFrameTransformTaskAction.Response(true));
237-
}
226+
buildTask.start(previousCheckpoint, listener);
238227
}
239228

240229
@Override

x-pack/plugin/data-frame/src/main/java/org/elasticsearch/xpack/dataframe/transforms/DataFrameTransformTask.java

Lines changed: 7 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -174,13 +174,8 @@ public long getInProgressCheckpoint() {
174174
}
175175
}
176176

177-
public boolean isStopped() {
178-
IndexerState currentState = getIndexer() == null ? initialIndexerState : getIndexer().getState();
179-
return currentState.equals(IndexerState.STOPPED);
180-
}
181-
182-
boolean isInitialRun() {
183-
return getIndexer() != null && getIndexer().initialRun();
177+
public void setTaskStateStopped() {
178+
taskState.set(DataFrameTransformTaskState.STOPPED);
184179
}
185180

186181
/**
@@ -235,11 +230,9 @@ public synchronized void start(Long startingCheckpoint, ActionListener<Response>
235230

236231
public synchronized void stop() {
237232
if (getIndexer() == null) {
238-
return;
239-
}
240-
// taskState is initialized as STOPPED and is updated in tandem with the indexerState
241-
// Consequently, if it is STOPPED, we consider the whole task STOPPED.
242-
if (taskState.get() == DataFrameTransformTaskState.STOPPED) {
233+
// If there is no indexer the task has not been triggered
234+
// but it still needs to be stopped and removed
235+
shutdown();
243236
return;
244237
}
245238

@@ -609,6 +602,8 @@ protected void onFinish(ActionListener<Void> listener) {
609602
protected void onStop() {
610603
auditor.info(transformConfig.getId(), "Indexer has stopped");
611604
logger.info("Data frame transform [{}] indexer has stopped", transformConfig.getId());
605+
606+
transformTask.setTaskStateStopped();
612607
transformsConfigManager.putOrUpdateTransformStats(
613608
new DataFrameTransformStateAndStats(transformId, transformTask.getState(), getStats(),
614609
DataFrameTransformCheckpointingInfo.EMPTY), // TODO should this be null

x-pack/plugin/src/test/resources/rest-api-spec/test/data_frame/transforms_start_stop.yml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -114,8 +114,8 @@ teardown:
114114
transform_id: "airline-transform-start-stop"
115115
- match: { count: 1 }
116116
- match: { transforms.0.id: "airline-transform-start-stop" }
117-
# - match: { transforms.0.state.indexer_state: "stopped" }
118-
# - match: { transforms.0.state.task_state: "stopped" }
117+
- match: { transforms.0.state.indexer_state: "stopped" }
118+
- match: { transforms.0.state.task_state: "stopped" }
119119

120120
- do:
121121
data_frame.start_data_frame_transform:

0 commit comments

Comments
 (0)