Fix raw type usage: replace Collections.EMPTY_LIST with Collections.emptyList()

geaflow/geaflow-common/src/main/java/org/apache/geaflow/common/serialize/impl/KryoSerializer.java

@@ -74,7 +74,7 @@ protected Kryo initialValue() {
             kryo.setRegistrationRequired(false);
 
             kryo.register(Arrays.asList("").getClass(), new ArraysAsListSerializer());
-            kryo.register(Collections.EMPTY_LIST.getClass(), new CollectionsEmptyListSerializer());
+            kryo.register(Collections.emptyList().getClass(), new CollectionsEmptyListSerializer());
             kryo.register(Collections.singletonList("").getClass(),
                 new CollectionsSingletonListSerializer());
             kryo.register(ClosureSerializer.Closure.class, new ClosureSerializer());

geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/org/apache/geaflow/cluster/web/handler/PipelineRestHandler.java

@@ -78,7 +78,7 @@ public ApiResponse<Collection<CycleMetrics>> queryCycleList(@PathParam("pipeline
             metricFetcher.update();
             PipelineMetricCache cache = metricCache.getPipelineMetricCaches().get(pipelineName);
             if (cache == null) {
-                return ApiResponse.success(Collections.EMPTY_LIST);
+                return ApiResponse.success(Collections.emptyList());
             }
             return ApiResponse.success(cache.getCycleMetricList().values());
         } catch (Throwable t) {

geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/org/apache/geaflow/cluster/web/HttpServerTest.java

@@ -60,7 +60,7 @@ public void run() {
             }
         }).start();
 
-        Mockito.when(clusterManager.getContainerInfos()).thenReturn(Collections.EMPTY_MAP);
+        Mockito.when(clusterManager.getContainerInfos()).thenReturn(Collections.emptyMap());
 
         latch.await();
         doGet("http://localhost:8090/", "rest/overview");

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-kafka/src/main/java/org/apache/geaflow/dsl/connector/kafka/KafkaTableSource.java

@@ -235,7 +235,7 @@ public <T> FetchData<T> fetch(Partition partition, Optional<Offset> startOffset,
         if (windowStartOffset == null || windowStartOffset.timestamp() >= windowEndTimeMs) {
             // no data in current window, skip!
             KafkaOffset offset = new KafkaOffset(partitionWithOffset.f1, windowEndTimeMs);
-            return (FetchData<T>) FetchData.createStreamFetch(Collections.EMPTY_LIST, offset, false);
+            return (FetchData<T>) FetchData.createStreamFetch(Collections.emptyList(), offset, false);
         }
         List<String> dataList = new ArrayList<>();
         long responseMaxTimestamp = -1;

geaflow/geaflow-plugins/geaflow-store/geaflow-store-memory/src/main/java/org/apache/geaflow/store/memory/StaticGraphMemoryCSRStore.java

@@ -218,7 +218,7 @@ public Map<K, Integer> getDict() {
 
         public List<IEdge<K, EV>> getEdges(K sid) {
             int pos = getDictId(sid);
-            return pos == NON_EXIST ? Collections.EMPTY_LIST : getEdges(sid, pos);
+            return pos == NON_EXIST ? Collections.emptyList() : getEdges(sid, pos);
         }
 
         public List<IEdge<K, EV>> getEdges(K sid, int pos) {

geaflow/geaflow-plugins/geaflow-store/geaflow-store-memory/src/main/java/org/apache/geaflow/store/memory/StaticGraphMemoryStore.java

@@ -71,7 +71,7 @@ protected IVertex<K, VV> getVertex(K sid) {
     @Override
     protected List<IEdge<K, EV>> getEdges(K sid) {
         Tuple<IVertex<K, VV>, List<IEdge<K, EV>>> v = map.get(sid);
-        return v == null ? Collections.EMPTY_LIST : v.f1;
+        return v == null ? Collections.emptyList() : v.f1;
     }
 
     @Override