[FLINK-18406] Add Nullable annotation to DualKeyLinkedMap.getValueByKeyA

[FLINK-18406] Add Nullable annotation to DualKeyLinkedMap.getValueByKeyB

[FLINK-18406] Add NullableAnnotation to DualKeyLinkedMap.getKeyAByKeyB

[FLINK-18406] Add Nullable annotation to DualKeyLinkedMap.getKeyBByKeyA

[FLINK-18406] Add Nullable annotation to DualKeyLinkeMap put, removeKeyA and removeKeyB

This closes apache#12738.

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DualKeyLinkedMap.java

@@ -20,6 +20,8 @@
 
 import org.apache.flink.api.java.tuple.Tuple2;
 
+import javax.annotation.Nullable;
+
 import java.util.AbstractCollection;
 import java.util.Collection;
 import java.util.HashMap;
@@ -58,6 +60,7 @@ int size() {
 		return aMap.size();
 	}
 
+	@Nullable
 	V getValueByKeyA(A aKey) {
 		final Tuple2<B, V> value = aMap.get(aKey);
 
@@ -68,6 +71,7 @@ V getValueByKeyA(A aKey) {
 		}
 	}
 
+	@Nullable
 	V getValueByKeyB(B bKey) {
 		final A aKey = bMap.get(bKey);
 
@@ -78,10 +82,12 @@ V getValueByKeyB(B bKey) {
 		}
 	}
 
+	@Nullable
 	A getKeyAByKeyB(B bKey) {
 		return bMap.get(bKey);
 	}
 
+	@Nullable
 	B getKeyBByKeyA(A aKey) {
 		final Tuple2<B, V> value = aMap.get(aKey);
 
@@ -92,6 +98,7 @@ B getKeyBByKeyA(A aKey) {
 		}
 	}
 
+	@Nullable
 	V put(A aKey, B bKey, V value) {
 		final V oldValue = getValueByKeyA(aKey);
 
@@ -125,6 +132,7 @@ boolean containsKeyB(B bKey) {
 		return bMap.containsKey(bKey);
 	}
 
+	@Nullable
 	V removeKeyA(A aKey) {
 		Tuple2<B, V> aValue = aMap.remove(aKey);
 
@@ -136,6 +144,7 @@ V removeKeyA(A aKey) {
 		}
 	}
 
+	@Nullable
 	V removeKeyB(B bKey) {
 		A aKey = bMap.remove(bKey);
 

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java

@@ -604,7 +604,7 @@ private void maybeRemapOrphanedAllocation(
 			final SlotRequestId requestIdOfAllocatedSlot = pendingRequests.getKeyAByKeyB(allocationIdOfSlot);
 			if (requestIdOfAllocatedSlot != null) {
 				final PendingRequest requestOfAllocatedSlot = pendingRequests.getValueByKeyA(requestIdOfAllocatedSlot);
-				requestOfAllocatedSlot.setAllocationId(allocationIdOfRequest);
+				checkNotNull(requestOfAllocatedSlot).setAllocationId(allocationIdOfRequest);
 
 				// this re-insertion of initiatedRequestId will not affect its original insertion order
 				pendingRequests.put(requestIdOfAllocatedSlot, allocationIdOfRequest, requestOfAllocatedSlot);
@@ -1088,6 +1088,7 @@ void add(SlotRequestId slotRequestId, AllocatedSlot allocatedSlot) {
 		 * @param allocationID The allocation id
 		 * @return The allocated slot, null if we can't find a match
 		 */
+		@Nullable
 		AllocatedSlot get(final AllocationID allocationID) {
 			return allocatedSlotsById.getValueByKeyA(allocationID);
 		}