[flink] Use ordered set in PojoToRowConverter to make test stable

Author: leonardBang

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/utils/PojoToRowConverter.java

@@ -22,7 +22,6 @@
 import com.alibaba.fluss.row.InternalRow;
 import com.alibaba.fluss.row.TimestampLtz;
 import com.alibaba.fluss.row.TimestampNtz;
-import com.alibaba.fluss.shaded.guava32.com.google.common.collect.Sets;
 import com.alibaba.fluss.types.DataType;
 import com.alibaba.fluss.types.DataTypeRoot;
 import com.alibaba.fluss.types.DecimalType;
@@ -42,7 +41,9 @@
 import java.time.LocalDateTime;
 import java.time.LocalTime;
 import java.time.OffsetDateTime;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Set;
 
@@ -74,27 +75,27 @@ public class PojoToRowConverter<T> {
     private static final Map<DataTypeRoot, Set<Class<?>>> SUPPORTED_TYPES = new HashMap<>();
 
     static {
-        SUPPORTED_TYPES.put(DataTypeRoot.BOOLEAN, Sets.newHashSet(Boolean.class, boolean.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.TINYINT, Sets.newHashSet(Byte.class, byte.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.SMALLINT, Sets.newHashSet(Short.class, short.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.INTEGER, Sets.newHashSet(Integer.class, int.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.BIGINT, Sets.newHashSet(Long.class, long.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.FLOAT, Sets.newHashSet(Float.class, float.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.DOUBLE, Sets.newHashSet(Double.class, double.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.BOOLEAN, orderedSet(Boolean.class, boolean.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.TINYINT, orderedSet(Byte.class, byte.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.SMALLINT, orderedSet(Short.class, short.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.INTEGER, orderedSet(Integer.class, int.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.BIGINT, orderedSet(Long.class, long.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.FLOAT, orderedSet(Float.class, float.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.DOUBLE, orderedSet(Double.class, double.class));
         SUPPORTED_TYPES.put(
-                DataTypeRoot.CHAR, Sets.newHashSet(String.class, Character.class, char.class));
+                DataTypeRoot.CHAR, orderedSet(String.class, Character.class, char.class));
         SUPPORTED_TYPES.put(
-                DataTypeRoot.STRING, Sets.newHashSet(String.class, Character.class, char.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.BINARY, Sets.newHashSet(byte[].class));
-        SUPPORTED_TYPES.put(DataTypeRoot.BYTES, Sets.newHashSet(byte[].class));
-        SUPPORTED_TYPES.put(DataTypeRoot.DECIMAL, Sets.newHashSet(BigDecimal.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.DATE, Sets.newHashSet(LocalDate.class));
-        SUPPORTED_TYPES.put(DataTypeRoot.TIME_WITHOUT_TIME_ZONE, Sets.newHashSet(LocalTime.class));
+                DataTypeRoot.STRING, orderedSet(String.class, Character.class, char.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.BINARY, orderedSet(byte[].class));
+        SUPPORTED_TYPES.put(DataTypeRoot.BYTES, orderedSet(byte[].class));
+        SUPPORTED_TYPES.put(DataTypeRoot.DECIMAL, orderedSet(BigDecimal.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.DATE, orderedSet(LocalDate.class));
+        SUPPORTED_TYPES.put(DataTypeRoot.TIME_WITHOUT_TIME_ZONE, orderedSet(LocalTime.class));
         SUPPORTED_TYPES.put(
-                DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE, Sets.newHashSet(LocalDateTime.class));
+                DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE, orderedSet(LocalDateTime.class));
         SUPPORTED_TYPES.put(
                 DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE,
-                Sets.newHashSet(Instant.class, OffsetDateTime.class));
+                orderedSet(Instant.class, OffsetDateTime.class));
         // Add more supported types as needed
 
     }
@@ -332,4 +333,19 @@ public GenericRow convert(T pojo) {
 
         return row;
     }
+
+    /**
+     * Utility method to create an ordered {@link LinkedHashSet} containing the specified Java type
+     * classes.
+     *
+     * <p>The returned set maintains the insertion order of the provided classes.
+     *
+     * @param javaTypes The Java type classes to include in the set. May be one or more classes.
+     * @return A new {@link LinkedHashSet} containing the given classes, preserving their order.
+     */
+    private static LinkedHashSet<Class<?>> orderedSet(Class<?>... javaTypes) {
+        LinkedHashSet<Class<?>> linkedHashSet = new LinkedHashSet<>();
+        linkedHashSet.addAll(Arrays.asList(javaTypes));
+        return linkedHashSet;
+    }
 }

fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/utils/PojoToRowConverterTest.java

@@ -234,7 +234,7 @@ public void testUnsupportedJavaClass() {
         assertThatThrownBy(() -> new PojoToRowConverter<>(ProductWithPrice.class, rowType))
                 .isInstanceOf(UnsupportedOperationException.class)
                 .hasMessageContaining(
-                        "Field Java type class java.lang.Long for field id is not supported, the supported Java types are [int, class java.lang.Integer]");
+                        "Field Java type class java.lang.Long for field id is not supported, the supported Java types are [class java.lang.Integer, int]");
     }
 
     @Test