ordered partition keys

Author: xiaochen-zhou

fluss-rpc/src/main/java/com/alibaba/fluss/rpc/util/CommonRpcMessageUtils.java

@@ -240,21 +240,30 @@ public static ResolvedPartitionSpec toResolvedPartitionSpec(PbPartitionSpec pbPa
     public static ResolvedPartitionSpec toResolvedPartitionSpec(
             PbPartitionSpec pbPartitionSpec, List<String> orderedPartitionKeys) {
 
-        Map<String, String> keyValueMap = new HashMap<>();
-        for (PbKeyValue pbKeyValue : pbPartitionSpec.getPartitionKeyValuesList()) {
-            keyValueMap.put(pbKeyValue.getKey(), pbKeyValue.getValue());
+        List<PbKeyValue> partitionKeyValuesList = pbPartitionSpec.getPartitionKeyValuesList();
+        if (partitionKeyValuesList.size() != orderedPartitionKeys.size()) {
+            return toResolvedPartitionSpec(pbPartitionSpec);
+        }
+
+        Map<String, String> pbkeyValueMap = new HashMap<>();
+        for (PbKeyValue pbKeyValue : partitionKeyValuesList) {
+            if (!orderedPartitionKeys.contains(pbKeyValue.getKey())) {
+                return toResolvedPartitionSpec(pbPartitionSpec);
+            }
+            pbkeyValueMap.put(pbKeyValue.getKey(), pbKeyValue.getValue());
         }
-        List<String> presentKeys = new ArrayList<>();
-        List<String> orderedValues = new ArrayList<>();
+
+        List<String> partitionKeys = new ArrayList<>();
+        List<String> orderedPartitionValues = new ArrayList<>();
 
         for (String orderedKey : orderedPartitionKeys) {
-            String value = keyValueMap.get(orderedKey);
+            String value = pbkeyValueMap.get(orderedKey);
             if (value != null) {
-                presentKeys.add(orderedKey);
-                orderedValues.add(value);
+                partitionKeys.add(orderedKey);
+                orderedPartitionValues.add(value);
             }
         }
 
-        return new ResolvedPartitionSpec(presentKeys, orderedValues);
+        return new ResolvedPartitionSpec(partitionKeys, orderedPartitionValues);
     }
 }