Revert "Refactor AllocatedPersistentTask#init(), move rollup logic out of ctor (#46288)"

This reverts commit d999942.

Author: polyfractal

server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java

@@ -18,7 +18,6 @@
  */
 package org.elasticsearch.persistent;
 
-import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.action.ActionListener;
@@ -38,13 +37,13 @@
  */
 public class AllocatedPersistentTask extends CancellableTask {
 
-    private static final Logger logger = LogManager.getLogger(AllocatedPersistentTask.class);
     private final AtomicReference<State> state;
 
     private volatile String persistentTaskId;
     private volatile long allocationId;
     private volatile @Nullable Exception failure;
     private volatile PersistentTasksService persistentTasksService;
+    private volatile Logger logger;
     private volatile TaskManager taskManager;
 
     public AllocatedPersistentTask(long id, String type, String action, String description, TaskId parentTask,
@@ -86,9 +85,10 @@ public String getPersistentTaskId() {
         return persistentTaskId;
     }
 
-    protected void init(PersistentTasksService persistentTasksService, TaskManager taskManager,
-                        String persistentTaskId, long allocationId) {
+    void init(PersistentTasksService persistentTasksService, TaskManager taskManager, Logger logger, String persistentTaskId, long
+            allocationId) {
         this.persistentTasksService = persistentTasksService;
+        this.logger = logger;
         this.taskManager = taskManager;
         this.persistentTaskId = persistentTaskId;
         this.allocationId = allocationId;

server/src/main/java/org/elasticsearch/persistent/PersistentTasksNodeService.java

@@ -183,7 +183,7 @@ public Task createTask(long id, String type, String action, TaskId parentTaskId,
 
         boolean processed = false;
         try {
-            task.init(persistentTasksService, taskManager, taskInProgress.getId(), taskInProgress.getAllocationId());
+            task.init(persistentTasksService, taskManager, logger, taskInProgress.getId(), taskInProgress.getAllocationId());
             logger.trace("Persistent task [{}] with id [{}] and allocation id [{}] was created", task.getAction(),
                     task.getPersistentTaskId(), task.getAllocationId());
             try {

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/job/RollupJobTask.java

@@ -22,9 +22,7 @@
 import org.elasticsearch.persistent.PersistentTaskState;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.persistent.PersistentTasksExecutor;
-import org.elasticsearch.persistent.PersistentTasksService;
 import org.elasticsearch.tasks.TaskId;
-import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.indexing.IndexerState;
@@ -152,59 +150,44 @@ protected void onAbort() {
     private final RollupJob job;
     private final SchedulerEngine schedulerEngine;
     private final ThreadPool threadPool;
-    private final Client client;
-    private final IndexerState initialIndexerState;
-    private final Map<String, Object> initialPosition;
-    private RollupIndexer indexer;
+    private final RollupIndexer indexer;
 
     RollupJobTask(long id, String type, String action, TaskId parentTask, RollupJob job, RollupJobStatus state,
                   Client client, SchedulerEngine schedulerEngine, ThreadPool threadPool, Map<String, String> headers) {
         super(id, type, action, RollupField.NAME + "_" + job.getConfig().getId(), parentTask, headers);
         this.job = job;
         this.schedulerEngine = schedulerEngine;
         this.threadPool = threadPool;
-        this.client = client;
-        if (state == null) {
-            this.initialIndexerState = null;
-            this.initialPosition = null;
-        } else {
-            this.initialIndexerState = state.getIndexerState();
-            this.initialPosition = state.getPosition();
-        }
-
-    }
 
-    @Override
-    protected void init(PersistentTasksService persistentTasksService, TaskManager taskManager,
-                        String persistentTaskId, long allocationId) {
-        super.init(persistentTasksService, taskManager, persistentTaskId, allocationId);
-
-        // If initial position is not null, we are resuming rather than starting fresh.
-        IndexerState indexerState = IndexerState.STOPPED;
-        if (initialPosition != null) {
-            logger.debug("We have existing state, setting state to [" + initialIndexerState + "] " +
-                "and current position to [" + initialPosition + "] for job [" + job.getConfig().getId() + "]");
-            if (initialIndexerState.equals(IndexerState.INDEXING)) {
+        // If status is not null, we are resuming rather than starting fresh.
+        Map<String, Object> initialPosition = null;
+        IndexerState initialState = IndexerState.STOPPED;
+        if (state != null) {
+            final IndexerState existingState = state.getIndexerState();
+            logger.debug("We have existing state, setting state to [" + existingState + "] " +
+                    "and current position to [" + state.getPosition() + "] for job [" + job.getConfig().getId() + "]");
+            if (existingState.equals(IndexerState.INDEXING)) {
                 /*
                  * If we were indexing, we have to reset back to STARTED otherwise the indexer will be "stuck" thinking
                  * it is indexing but without the actual indexing thread running.
                  */
-                indexerState = IndexerState.STARTED;
+                initialState = IndexerState.STARTED;
 
-            } else if (initialIndexerState.equals(IndexerState.ABORTING) || initialIndexerState.equals(IndexerState.STOPPING)) {
+            } else if (existingState.equals(IndexerState.ABORTING) || existingState.equals(IndexerState.STOPPING)) {
                 // It shouldn't be possible to persist ABORTING, but if for some reason it does,
                 // play it safe and restore the job as STOPPED.  An admin will have to clean it up,
                 // but it won't be running, and won't delete itself either.  Safest option.
                 // If we were STOPPING, that means it persisted but was killed before finally stopped... so ok
                 // to restore as STOPPED
-                indexerState = IndexerState.STOPPED;
+                initialState = IndexerState.STOPPED;
             } else  {
-                indexerState = initialIndexerState;
+                initialState = existingState;
             }
+            initialPosition = state.getPosition();
 
         }
-        this.indexer = new ClientRollupPageManager(job, indexerState, initialPosition,
-            new ParentTaskAssigningClient(client, getParentTaskId()));
+        this.indexer = new ClientRollupPageManager(job, initialState, initialPosition,
+                new ParentTaskAssigningClient(client, new TaskId(getPersistentTaskId())));
     }
 
     @Override