fixup! [FLINK-35652][table] Shuffle input stream of lookup join based on LOOKUP hint

Author: reswqa

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/connector/source/LookupRuntimeProviderContext.java

@@ -37,11 +37,6 @@ public final class LookupRuntimeProviderContext implements LookupTableSource.Loo
 
     private final boolean preferCustomShuffle;
 
-    public LookupRuntimeProviderContext(int[][] lookupKeys) {
-        this.lookupKeys = lookupKeys;
-        this.preferCustomShuffle = false;
-    }
-
     public LookupRuntimeProviderContext(int[][] lookupKeys, boolean preferCustomShuffle) {
         this.lookupKeys = lookupKeys;
         this.preferCustomShuffle = preferCustomShuffle;

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/partitioner/RowDataCustomStreamPartitioner.java

@@ -50,7 +50,9 @@ public int selectChannel(SerializationDelegate<StreamRecord<RowData>> record) {
                     "Could not extract key from " + record.getInstance().getValue(), e);
         }
         int partition = partitioner.partition(key, numberOfChannels);
-        Preconditions.checkState(partition < numberOfChannels);
+        Preconditions.checkState(
+                partition < numberOfChannels,
+                "The partition computed by custom partitioner is out of range, please check the logic of custom partitioner.");
         return partition;
     }