[Transform] improve transform log/audit correctness when stopped (#99917)

move auditor and logging for transform being stopped, so it gets reported after
the p-task has been marked for completion

docs/changelog/99917.yaml

@@ -0,0 +1,5 @@
+pr: 99917
+summary: Improve transform log/audit correctness when stopped
+area: Transform
+type: enhancement
+issues: []

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/ClientTransformIndexer.java

@@ -395,6 +395,7 @@ class ClientTransformIndexer extends TransformIndexer {
 
     @Override
     protected void onStop() {
+        logger.debug(() -> format("[%s] transform initiating stop", transformConfig.getId()));
         closePointInTime();
         super.onStop();
     }

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformIndexer.java

@@ -564,7 +564,10 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
 
     @Override
     protected void afterFinishOrFailure() {
-        finishIndexerThreadShutdown();
+        finishIndexerThreadShutdown(() -> {
+            auditor.info(transformConfig.getId(), "Transform has stopped.");
+            logger.info("[{}] transform has stopped.", transformConfig.getId());
+        });
     }
 
     @Override
@@ -646,12 +649,6 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
         }
     }
 
-    @Override
-    protected void onStop() {
-        auditor.info(transformConfig.getId(), "Transform has stopped.");
-        logger.info("[{}] transform has stopped.", transformConfig.getId());
-    }
-
     @Override
     protected void onAbort() {
         auditor.info(transformConfig.getId(), "Received abort request, stopping transform.");
@@ -1203,7 +1200,7 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
         }
     }
 
-    private void finishIndexerThreadShutdown() {
+    private void finishIndexerThreadShutdown(Runnable next) {
         synchronized (context) {
             indexerThreadShuttingDown = false;
             if (saveStateRequestedDuringIndexerThreadShutdown) {
@@ -1212,7 +1209,9 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
                 if (context.shouldStopAtCheckpoint() && nextCheckpoint == null) {
                     stop();
                 }
-                doSaveState(getState(), getPosition(), () -> {});
+                doSaveState(getState(), getPosition(), next);
+            } else {
+                next.run();
             }
         }
     }