Passing parent task id to the refresh request within reindex data str…

…eams (#121533) (#121605)
elastic · Feb 4, 2025 · 2d7d6fe · 2d7d6fe
1 parent facce5f
commit 2d7d6fe
Showing 1 changed file with 17 additions and 19 deletions.
diff --git a/...in/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java b/...in/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexTransportAction.java
@@ -368,26 +368,24 @@ private void sanityCheck(
     ) {
         if (Assertions.ENABLED) {
             logger.debug("Comparing source [{}] and dest [{}] doc counts", sourceIndexName, destIndexName);
-            client.execute(
-                RefreshAction.INSTANCE,
-                new RefreshRequest(destIndexName),
-                listener.delegateFailureAndWrap((delegate, ignored) -> {
-                    getIndexDocCount(sourceIndexName, parentTaskId, delegate.delegateFailureAndWrap((delegate1, sourceCount) -> {
-                        getIndexDocCount(destIndexName, parentTaskId, delegate1.delegateFailureAndWrap((delegate2, destCount) -> {
-                            assert Objects.equals(sourceCount, destCount)
-                                : String.format(
-                                    Locale.ROOT,
-                                    "source index [%s] has %d docs and dest [%s] has %d docs",
-                                    sourceIndexName,
-                                    sourceCount,
-                                    destIndexName,
-                                    destCount
-                                );
-                            delegate2.onResponse(null);
-                        }));
+            RefreshRequest refreshRequest = new RefreshRequest(destIndexName);
+            refreshRequest.setParentTask(parentTaskId);
+            client.execute(RefreshAction.INSTANCE, refreshRequest, listener.delegateFailureAndWrap((delegate, ignored) -> {
+                getIndexDocCount(sourceIndexName, parentTaskId, delegate.delegateFailureAndWrap((delegate1, sourceCount) -> {
+                    getIndexDocCount(destIndexName, parentTaskId, delegate1.delegateFailureAndWrap((delegate2, destCount) -> {
+                        assert Objects.equals(sourceCount, destCount)
+                            : String.format(
+                                Locale.ROOT,
+                                "source index [%s] has %d docs and dest [%s] has %d docs",
+                                sourceIndexName,
+                                sourceCount,
+                                destIndexName,
+                                destCount
+                            );
+                        delegate2.onResponse(null);
                     }));
-                })
-            );
+                }));
+            }));
         } else {
             listener.onResponse(null);
         }