diff --git a/x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/xpack/ml/integration/BasicDistributedJobsIT.java b/x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/xpack/ml/integration/BasicDistributedJobsIT.java
index a6eecfeb94c32..aee2f972b5bc4 100644
--- a/x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/xpack/ml/integration/BasicDistributedJobsIT.java
+++ b/x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/xpack/ml/integration/BasicDistributedJobsIT.java
@@ -93,7 +93,6 @@ public void testFailOverBasics() throws Exception {
         assertRecentLastTaskStateChangeTime(MlTasks.jobTaskId(job.getId()), Duration.of(10, ChronoUnit.SECONDS), null);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/82591")
     public void testFailOverBasics_withDataFeeder() throws Exception {
         internalCluster().ensureAtLeastNumDataNodes(4);
         ensureStableCluster(4);
@@ -166,6 +165,12 @@ public void testFailOverBasics_withDataFeeder() throws Exception {
             assertEquals(1, statsResponse.getResponse().results().size());
             assertEquals(DatafeedState.STARTED, statsResponse.getResponse().results().get(0).getDatafeedState());
         });
+
+        // Ensure all nodes have a consistent view of the cluster state before cleanup
+        // runs, otherwise the stop-datafeed request during cleanup may be routed to a
+        // node that hasn't yet received the state update showing the datafeed as started.
+        // See https://github.com/elastic/elasticsearch/issues/82591
+        waitNoPendingTasksOnAll();
     }
 
     public void testJobAutoClose() throws Exception {