Fix tests for new markAsLocallyAborted method (#75021)

Two test fixes following on from #74115

Fixes #75004
Fixes #75012

Author: droberts195

server/src/internalClusterTest/java/org/elasticsearch/persistent/PersistentTasksExecutorIT.java

@@ -8,7 +8,6 @@
 
 package org.elasticsearch.persistent;
 
-import org.apache.lucene.util.LuceneTestCase;
 import org.elasticsearch.ResourceAlreadyExistsException;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.support.PlainActionFuture;
@@ -34,6 +33,7 @@
 import java.util.Objects;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertFutureThrows;
+import static org.hamcrest.Matchers.either;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -42,7 +42,6 @@
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.core.Is.is;
 
-@LuceneTestCase.AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/75012")
 @ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.SUITE, minNumDataNodes = 2)
 public class PersistentTasksExecutorIT extends ESIntegTestCase {
 
@@ -363,7 +362,13 @@ public void testAbortLocally() throws Exception {
             // Verify that the task is STILL in internal cluster state, unassigned, with a reason indicating local abort
             PersistentTask<?> task = assertClusterStateHasTask(taskId);
             assertThat(task.getAssignment().getExecutorNode(), nullValue());
-            assertThat(task.getAssignment().getExplanation(), equalTo("Simulating local abort"));
+            // Although the assignment explanation is initially set to "Simulating local abort", because
+            // of the way we prevent reassignment to the same node in this test it may quickly change to
+            // "non cluster state condition prevents assignment" - either proves the unassignment worked
+            assertThat(task.getAssignment().getExplanation(),
+                either(equalTo("Simulating local abort"))
+                    .or(equalTo("non cluster state condition prevents assignment"))
+            );
         });
 
         // Allow it to be reassigned again

server/src/test/java/org/elasticsearch/persistent/PersistentTasksNodeServiceTests.java

@@ -318,7 +318,6 @@ public boolean isLocalAbortSupported() {
         assertThat(taskManager.getTasks().values(), empty());
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/75004")
     public void testTaskLocalAbort() {
         AtomicReference<String> capturedTaskId = new AtomicReference<>();
         AtomicReference<ActionListener<PersistentTask<?>>> capturedListener = new AtomicReference<>();
@@ -407,7 +406,10 @@ public boolean isLocalAbortSupported() {
                     equalTo("attempt to fail task [test] with id [" + persistentId + "] which has been locally aborted"));
                 break;
             case 2:
-                runningTask.markAsLocallyAborted("second local abort");
+                IllegalStateException e2 = expectThrows(IllegalStateException.class,
+                    () -> runningTask.markAsLocallyAborted("second local abort"));
+                assertThat(e2.getMessage(),
+                    equalTo("attempt to locally abort task [test] with id [" + persistentId + "] which has already been locally aborted"));
                 break;
         }