[FLINK-18332][state] Add error message to precondition in KeyGroupPar…

…titionedPriorityQueue

This closes apache#12508.

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/KeyGroupPartitionedPriorityQueue.java

@@ -36,6 +36,8 @@
 import java.util.HashSet;
 import java.util.Set;
 
+import static org.apache.flink.util.Preconditions.checkArgument;
+
 /**
  * This implementation of {@link InternalPriorityQueue} is internally partitioned into sub-queues per key-group and
  * essentially works as a heap-of-heaps. Instances will have set semantics for elements if the sub-queues have set
@@ -181,7 +183,11 @@ private int computeKeyGroupIndex(T element) {
 	}
 
 	private int globalKeyGroupToLocalIndex(int keyGroupId) {
-		return keyGroupId - firstKeyGroup;
+		int keyGroupIndex = keyGroupId - firstKeyGroup;
+		checkArgument((keyGroupIndex >= 0 && keyGroupIndex < keyGroupedHeaps.length),
+			"key group from %s to %s does not contain %s", firstKeyGroup,
+			(firstKeyGroup + keyGroupedHeaps.length), keyGroupId);
+		return keyGroupIndex;
 	}
 
 	@Nonnull