[FLINK-12039] Remove ASSIGNED_SLOT_UPDATER in Execution.tryAssignReso…

…urce

This closes apache#10168 .

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java

@@ -121,11 +121,6 @@ public class Execution implements AccessExecution, Archiveable<ArchivedExecution
 	private static final AtomicReferenceFieldUpdater<Execution, ExecutionState> STATE_UPDATER =
 			AtomicReferenceFieldUpdater.newUpdater(Execution.class, ExecutionState.class, "state");
 
-	private static final AtomicReferenceFieldUpdater<Execution, LogicalSlot> ASSIGNED_SLOT_UPDATER = AtomicReferenceFieldUpdater.newUpdater(
-		Execution.class,
-		LogicalSlot.class,
-		"assignedResource");
-
 	private static final Logger LOG = ExecutionGraph.LOG;
 
 	private static final int NUM_CANCEL_CALL_TRIES = 3;
@@ -304,7 +299,8 @@ public boolean tryAssignResource(final LogicalSlot logicalSlot) {
 		// only allow to set the assigned resource in state SCHEDULED or CREATED
 		// note: we also accept resource assignment when being in state CREATED for testing purposes
 		if (state == SCHEDULED || state == CREATED) {
-			if (ASSIGNED_SLOT_UPDATER.compareAndSet(this, null, logicalSlot)) {
+			if (assignedResource == null) {
+				assignedResource = logicalSlot;
 				if (logicalSlot.tryAssignPayload(this)) {
 					// check for concurrent modification (e.g. cancelling call)
 					if ((state == SCHEDULED || state == CREATED) && !taskManagerLocationFuture.isDone()) {
@@ -313,11 +309,11 @@ public boolean tryAssignResource(final LogicalSlot logicalSlot) {
 						return true;
 					} else {
 						// free assigned resource and return false
-						ASSIGNED_SLOT_UPDATER.set(this, null);
+						assignedResource = null;
 						return false;
 					}
 				} else {
-					ASSIGNED_SLOT_UPDATER.set(this, null);
+					assignedResource = null;
 					return false;
 				}
 			} else {