[Transform] Close PIT on abort (#143889)

When we force stop a transform, onAbort is eventually called to
immediately halt the progress of the checkpoint. onAbort now closes the
now unused PIT requests. This avoids the keepalive timeout and lets
Elasticsearch release the held segments. This largely only impacts our
integration tests that will not wait for that 30 seconds.

Author: prwhelan

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

@@ -486,6 +486,11 @@ protected void onStop() {
         closePointInTime(super::onStop);
     }
 
+    @Override
+    protected void onAbort() {
+        closePointInTime(super::onAbort);
+    }
+
     // visible for testing
     void closePointInTime(Runnable runAfter) {
         // we shouldn't need to do this, because a transform is only ever running on one thread anyway, but now that we're waiting for

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/ClientTransformIndexerTests.java

@@ -218,6 +218,30 @@ public void testPitInjection() throws InterruptedException {
         }
     }
 
+    public void testPitClosedOnAbort() throws InterruptedException {
+        TransformConfig config = new TransformConfig.Builder(TransformConfigTests.randomTransformConfig()).setSettings(
+            new SettingsConfig.Builder().setUsePit(true).build()
+        ).build();
+
+        try (var threadPool = createThreadPool()) {
+            final var client = new PitMockClient(threadPool, true);
+            MockClientTransformIndexer indexer = createMockIndexerForPitTest(client, config);
+
+            this.<SearchResponse>assertAsync(listener -> indexer.doNextSearch(0, listener), response -> {
+                assertThat(response.pointInTimeId(), equalBytes(new BytesArray("the_pit_id+")));
+            });
+
+            assertEquals(1L, client.getPitContextCounter());
+
+            indexer.onAbort();
+            assertEquals(0L, client.getPitContextCounter());
+
+            // calling onAbort again should be a no-op
+            indexer.onAbort();
+            assertEquals(0L, client.getPitContextCounter());
+        }
+    }
+
     public void testPitInjectionIfPitNotSupported() throws InterruptedException {
         // pit must be enabled, otherwise take a random config
         TransformConfig config = new TransformConfig.Builder(TransformConfigTests.randomTransformConfig()).setSettings(