[hotfix] Cleanup raw types around PriorityQueueSetFactory

tongcheng-elong · Feb 24, 2021 · 7f3aa39 · 7f3aa39
1 parent 2f16bff
commit 7f3aa39
Show file tree

Hide file tree

Showing 7 changed files with 11 additions and 11 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/PriorityQueueSetFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/PriorityQueueSetFactory.java
@@ -36,7 +36,7 @@ public interface PriorityQueueSetFactory {
  * @return the queue with the specified unique name.
  */
  @Nonnull
- <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  KeyGroupedInternalPriorityQueue<T> create(
  @Nonnull String stateName,
  @Nonnull TypeSerializer<T> byteOrderedElementSerializer);

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
@@ -157,13 +157,13 @@ public HeapKeyedStateBackend(
  @SuppressWarnings("unchecked")
  @Nonnull
  @Override
- public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  KeyGroupedInternalPriorityQueue<T> create(
  @Nonnull String stateName,
  @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {
 
- final HeapPriorityQueueSnapshotRestoreWrapper existingState =
- registeredPQStates.get(stateName);
+ final HeapPriorityQueueSnapshotRestoreWrapper<T> existingState =
+ (HeapPriorityQueueSnapshotRestoreWrapper<T>) registeredPQStates.get(stateName);
 
  if (existingState != null) {
  // TODO we implement the simple way of supporting the current functionality, mimicking
@@ -197,7 +197,7 @@ KeyGroupedInternalPriorityQueue<T> create(
  }
 
  @Nonnull
- private <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ private <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  KeyGroupedInternalPriorityQueue<T> createInternal(
  RegisteredPriorityQueueStateBackendMetaInfo<T> metaInfo) {
 

diff --git a/...untime/src/main/java/org/apache/flink/runtime/state/heap/HeapPriorityQueueSetFactory.java b/...untime/src/main/java/org/apache/flink/runtime/state/heap/HeapPriorityQueueSetFactory.java
@@ -50,7 +50,7 @@ public HeapPriorityQueueSetFactory(
 
  @Nonnull
  @Override
- public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  HeapPriorityQueueSet<T> create(
  @Nonnull String stateName,
  @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {

diff --git a/...-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java b/...-runtime/src/test/java/org/apache/flink/runtime/state/ttl/mock/MockKeyedStateBackend.java
@@ -278,7 +278,7 @@ private static <K> void copyEntry(
 
  @Nonnull
  @Override
- public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  KeyGroupedInternalPriorityQueue<T> create(
  @Nonnull String stateName,
  @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {

diff --git a/...ksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java b/...ksdb/src/main/java/org/apache/flink/contrib/streaming/state/RocksDBKeyedStateBackend.java
@@ -455,7 +455,7 @@ public void dispose() {
 
  @Nonnull
  @Override
- public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  KeyGroupedInternalPriorityQueue<T> create(
  @Nonnull String stateName,
  @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {

diff --git a/...rc/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java b/...rc/main/java/org/apache/flink/contrib/streaming/state/RocksDBPriorityQueueSetFactory.java
@@ -98,7 +98,7 @@ public class RocksDBPriorityQueueSetFactory implements PriorityQueueSetFactory {
 
  @Nonnull
  @Override
- public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  KeyGroupedInternalPriorityQueue<T> create(
  @Nonnull String stateName,
  @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {

diff --git a/...rg/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java b/...rg/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java
@@ -244,8 +244,8 @@ private <N, SV, S extends State, IS extends S> IS createState(
 
  @Nonnull
  @Override
- @SuppressWarnings({"rawtypes", "unchecked"})
- public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed>
+ @SuppressWarnings({"unchecked"})
+ public <T extends HeapPriorityQueueElement & PriorityComparable<? super T> & Keyed<?>>
  KeyGroupedInternalPriorityQueue<T> create(
  @Nonnull String stateName,
  @Nonnull TypeSerializer<T> byteOrderedElementSerializer) {