[Transform] Execute _refresh separately from DBQ, with system permissions. (#88005)

Author: przemekwitek

docs/changelog/88005.yaml

@@ -0,0 +1,6 @@
+pr: 88005
+summary: "Execute `_refresh` separately from DBQ, with system permissions"
+area: Transform
+type: bug
+issues:
+ - 88001

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

@@ -72,13 +72,12 @@ static DeleteByQueryRequest buildDeleteByQueryRequest(
         /* other dbq options not set and why:
          * - timeout: we do not timeout for search, so we don't timeout for dbq
          * - batch size: don't use max page size search, dbq should be simple
+         * - refresh: we call refresh separately, after DBQ is executed because refresh should be executed with system permissions
          */
         request.setSlices(AbstractBulkByScrollRequest.AUTO_SLICES)
             .setBatchSize(AbstractBulkByScrollRequest.DEFAULT_SCROLL_SIZE)
             // this should not happen, but still go over version conflicts and report later
             .setAbortOnVersionConflict(false)
-            // refresh the index, so docs are really gone
-            .setRefresh(true)
             // use transforms retry mechanics instead
             .setMaxRetries(0)
             .indices(destConfig.getIndex());

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

@@ -463,6 +463,11 @@ private void executeRetentionPolicy(ActionListener<Void> listener) {
         );
         getStats().markStartDelete();
 
+        ActionListener<RefreshResponse> deleteByQueryAndRefreshDoneListener = ActionListener.wrap(
+            refreshResponse -> finalizeCheckpoint(listener),
+            listener::onFailure
+        );
+
         doDeleteByQuery(deleteByQuery, ActionListener.wrap(bulkByScrollResponse -> {
             logger.trace(() -> format("[%s] dbq response: [%s]", getJobId(), bulkByScrollResponse));
 
@@ -493,7 +498,9 @@ private void executeRetentionPolicy(ActionListener<Void> listener) {
                 return;
             }
 
-            finalizeCheckpoint(listener);
+            // Since we configure DBQ request *not* to perform a refresh, we need to perform the refresh manually.
+            // This separation ensures that the DBQ runs with user permissions and the refresh runs with system permissions.
+            refreshDestinationIndex(deleteByQueryAndRefreshDoneListener);
         }, listener::onFailure));
     }
 

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

@@ -77,7 +77,7 @@ public void testTimeBasedRetentionPolicyConfig() {
         assertNotNull(deleteByQueryRequest.getSearchRequest().source());
         assertNotNull(deleteByQueryRequest.getSearchRequest().source().query());
 
-        assertTrue(deleteByQueryRequest.isRefresh());
+        assertFalse(deleteByQueryRequest.isRefresh());
         assertEquals(0, deleteByQueryRequest.getMaxRetries());
         assertEquals(1, deleteByQueryRequest.getSearchRequest().indices().length);
         assertEquals(destConfig.getIndex(), deleteByQueryRequest.getSearchRequest().indices()[0]);