Merge branch 'develop' into 'feature/dos'feature/dos

# Conflicts:
#   config/udf.plugins

14 files changed, 603 insertions, 73 deletions

diff --git a/groot-core/pom.xml b/groot-core/pom.xml
index e526024..322f63d 100644
--- a/groot-core/pom.xml
+++ b/groot-core/pom.xml
@@ -15,8 +15,8 @@
         <dependency>
             <groupId>com.fasterxml.uuid</groupId>
             <artifactId>java-uuid-generator</artifactId>
-            <version>5.1.0</version>
         </dependency>
+
         <dependency>
             <groupId>com.uber</groupId>
             <artifactId>h3</artifactId>
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/knowlegdebase/handler/AsnKnowledgeBaseHandler.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/knowlegdebase/handler/AsnKnowledgeBaseHandler.java
index a1927db..a0b9ce5 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/knowlegdebase/handler/AsnKnowledgeBaseHandler.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/knowlegdebase/handler/AsnKnowledgeBaseHandler.java
@@ -117,6 +117,7 @@ public class AsnKnowledgeBaseHandler extends AbstractKnowledgeBaseHandler {
                 }
 
             } catch (Exception e) {
+                log.error("Current class path {}", this.getClass().getProtectionDomain().getCodeSource().getLocation().getPath());
                 log.error("File {} operation failed. {} ", knowledgeBaseConfig.getFiles().get(i), e.getMessage());
                 return false;
             }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/NameSpaceType.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/NameSpaceType.java
deleted file mode 100644
index 1f6fd85..0000000
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/NameSpaceType.java
+++ /dev/null
@@ -1,30 +0,0 @@
-package com.geedgenetworks.core.udf.uuid;
-
-import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
-
-import java.util.UUID;
-
-import static com.geedgenetworks.common.exception.CommonErrorCode.ILLEGAL_ARGUMENT;
-
-public enum NameSpaceType {
-
-    NAMESPACE_IP("NAMESPACE_IP",UUID.fromString("6ba7b890-9dad-11d1-80b4-00c04fd430c8")),
-    NAMESPACE_DOMAIN("NAMESPACE_DOMAIN", UUID.fromString("6ba7b891-9dad-11d1-80b4-00c04fd430c8")),
-    NAMESPACE_APP("NAMESPACE_APP", UUID.fromString("6ba7b892-9dad-11d1-80b4-00c04fd430c8")),
-    NAMESPACE_SUBSCRIBER("NAMESPACE_SUBSCRIBER", UUID.fromString("6ba7b893-9dad-11d1-80b4-00c04fd430c8"));
-    private final String name;
-    private final UUID uuid;
-    NameSpaceType(String name, UUID uuid) {
-        this.name = name;
-        this.uuid = uuid;
-    }
-    public static UUID getUuidByName(String name) {
-        for (NameSpaceType nameSpaceType : NameSpaceType.values()) {
-            if (nameSpaceType.name.equals(name)) {
-                return nameSpaceType.uuid;
-            }
-        }
-        throw new GrootStreamRuntimeException(ILLEGAL_ARGUMENT,"No enum constant " + NameSpaceType.class.getCanonicalName() + "." + name);
-    }
-
-}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/Uuid.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUID.java
index 2c77108..1ce65bc 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/Uuid.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUID.java
@@ -11,7 +11,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.api.common.functions.RuntimeContext;
 
 @Slf4j
-public class Uuid implements ScalarFunction {
+public class UUID implements ScalarFunction {
     private String outputFieldName;
     private RandomBasedGenerator randomBasedGenerator;
     @Override
@@ -42,7 +42,6 @@ public class Uuid implements ScalarFunction {
 
     @Override
     public void close() {
-
     }
 
 
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDNameSpace.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDNameSpace.java
new file mode 100644
index 0000000..a8941e2
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDNameSpace.java
@@ -0,0 +1,43 @@
+package com.geedgenetworks.core.udf.uuid;
+
+import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import static com.geedgenetworks.common.exception.CommonErrorCode.ILLEGAL_ARGUMENT;
+
+public enum UUIDNameSpace {
+
+    NAMESPACE_IP(UUID.fromString("6ba7b890-9dad-11d1-80b4-00c04fd430c8")),
+    NAMESPACE_DOMAIN(UUID.fromString("6ba7b891-9dad-11d1-80b4-00c04fd430c8")),
+    NAMESPACE_APP(UUID.fromString("6ba7b892-9dad-11d1-80b4-00c04fd430c8")),
+    NAMESPACE_SUBSCRIBER(UUID.fromString("6ba7b893-9dad-11d1-80b4-00c04fd430c8"));
+
+    private final UUID uuid;
+
+    // Static map to hold the mapping from name to UUID
+    private static final Map<String, UUID> NAME_TO_UUID_MAP = new HashMap<>();
+
+    // Static block to populate the map
+    static {
+        for (UUIDNameSpace namespace : UUIDNameSpace.values()) {
+            NAME_TO_UUID_MAP.put(namespace.name(), namespace.uuid);
+        }
+    }
+
+    UUIDNameSpace(UUID uuid) {
+        this.uuid = uuid;
+    }
+
+    public static UUID getUUID(String name) {
+        UUID uuid = NAME_TO_UUID_MAP.get(name);
+        if (uuid == null) {
+            throw new GrootStreamRuntimeException(ILLEGAL_ARGUMENT,"No enum constant " + UUIDNameSpace.class.getCanonicalName() + "." + name);
+        }
+        return uuid;
+    }
+
+
+}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UuidV5.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDv5.java
index ad46ec4..b4ad808 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UuidV5.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDv5.java
@@ -13,39 +13,46 @@ import org.apache.flink.api.common.functions.RuntimeContext;
 import java.util.List;
 
 @Slf4j
-public class UuidV5 implements ScalarFunction {
+public class UUIDv5 implements ScalarFunction {
     private List<String> lookupFieldNames;
     private String outputFieldName;
     private NameBasedGenerator nameBasedGenerator;
+    private static final String NAMESPACE_KEY = "namespace";
     @Override
     public void open(RuntimeContext runtimeContext, UDFContext udfContext) {
 
-        if(udfContext.getOutput_fields()==null || udfContext.getParameters()==null || udfContext.getLookup_fields()==null){
+        if(udfContext.getOutput_fields() == null || udfContext.getParameters() == null || udfContext.getLookup_fields() == null){
             throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Missing required parameters");
         }
         if(udfContext.getOutput_fields().size() != 1){
             throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function output fields only support 1 value");
         }
-        if(!udfContext.getParameters().containsKey("namespace") ){
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "parameters must containkey namespace");
+        if(!udfContext.getParameters().containsKey(NAMESPACE_KEY) ){
+            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Parameters must contain key: " + NAMESPACE_KEY);
         }
+
         this.outputFieldName = udfContext.getOutput_fields().get(0);
         this.lookupFieldNames = udfContext.getLookup_fields();
-        this.nameBasedGenerator = Generators.nameBasedGenerator(NameSpaceType.getUuidByName(udfContext.getParameters().get("namespace").toString()));
+        String namespace = udfContext.getParameters().get(NAMESPACE_KEY).toString();
+        this.nameBasedGenerator = Generators.nameBasedGenerator(UUIDNameSpace.getUUID(namespace));
     }
 
     @Override
     public Event evaluate(Event event) {
-        StringBuilder sb = new StringBuilder();
-        for (int i = 0; i < lookupFieldNames.size(); i++) {
-            sb.append(event.getExtractedFields().getOrDefault(lookupFieldNames.get(i), ""));
-            if (i < lookupFieldNames.size() - 1) {
-                sb.append("_");
-            }
-        }
-        event.getExtractedFields()
-                .put(outputFieldName, nameBasedGenerator.generate(sb.toString()).toString());
+
+        String concatenatedFields = String.join("_",
+                lookupFieldNames.stream()
+                        .map(field -> event.getExtractedFields().getOrDefault(field, ""))
+                        .toArray(String[]::new)
+        );
+
+        // Generate the UUID based on concatenated fields
+        String generatedUUID = nameBasedGenerator.generate(concatenatedFields).toString();
+
+        // Set the generated UUID in the output field
+        event.getExtractedFields().put(outputFieldName, generatedUUID);
         return event;
+
     }
 
     @Override
@@ -57,4 +64,5 @@ public class UuidV5 implements ScalarFunction {
     public void close() {
 
     }
+
 }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UuidV7.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDv7.java
index 9dfbce3..49025ef 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UuidV7.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/uuid/UUIDv7.java
@@ -1,7 +1,6 @@
 package com.geedgenetworks.core.udf.uuid;
 
 import com.fasterxml.uuid.Generators;
-import com.fasterxml.uuid.impl.NameBasedGenerator;
 import com.fasterxml.uuid.impl.TimeBasedEpochGenerator;
 import com.geedgenetworks.common.Event;
 import com.geedgenetworks.common.exception.CommonErrorCode;
@@ -12,7 +11,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.api.common.functions.RuntimeContext;
 
 @Slf4j
-public class UuidV7 implements ScalarFunction {
+public class UUIDv7 implements ScalarFunction {
     private String outputFieldName;
     private TimeBasedEpochGenerator timeBasedEpochRandomGenerator;
 
@@ -44,6 +43,5 @@ public class UuidV7 implements ScalarFunction {
 
     @Override
     public void close() {
-
     }
 }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/LoadIntervalDataOptions.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/LoadIntervalDataOptions.java
new file mode 100644
index 0000000..a81794d
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/LoadIntervalDataOptions.java
@@ -0,0 +1,80 @@
+package com.geedgenetworks.core.utils;

+

+import java.io.Serializable;

+

+public class LoadIntervalDataOptions implements Serializable {

+    final String name;

+

+    final long intervalMs;

+    final boolean failOnException;

+    final boolean updateDataOnStart;

+

+    /**

+     * @param name 名称, 用于日志打印以及线程名称标识

+     * @param intervalMs 每隔多长时间更新数据

+     * @param failOnException 更新数据时发生异常是否失败(默认false), 为true时如果发现异常data()方法下次返回数据时会抛出异常

+     * @param updateDataOnStart start时是否先更新数据(默认true), 为false时start候intervalMs时间后才会第一个更新数据

+     */

+    private LoadIntervalDataOptions(String name, long intervalMs, boolean failOnException, boolean updateDataOnStart) {

+        this.name = name;

+        this.intervalMs = intervalMs;

+        this.failOnException = failOnException;

+        this.updateDataOnStart = updateDataOnStart;

+    }

+

+    public String getName() {

+        return name;

+    }

+

+    public long getIntervalMs() {

+        return intervalMs;

+    }

+

+    public boolean isFailOnException() {

+        return failOnException;

+    }

+

+    public boolean isUpdateDataOnStart() {

+        return updateDataOnStart;

+    }

+

+    public static Builder builder() {

+        return new Builder();

+    }

+

+    public static LoadIntervalDataOptions defaults(String name, long intervalMs) {

+        return builder().withName(name).withIntervalMs(intervalMs).build();

+    }

+

+    public static final class Builder {

+        private String name = "";

+        private long intervalMs = 1000 * 60 * 10;

+        private boolean failOnException = false;

+        private boolean updateDataOnStart = true;

+

+        public Builder withName(String name) {

+            this.name = name;

+            return this;

+        }

+

+        public Builder withIntervalMs(long intervalMs) {

+            this.intervalMs = intervalMs;

+            return this;

+        }

+

+        public Builder withFailOnException(boolean failOnException) {

+            this.failOnException = failOnException;

+            return this;

+        }

+

+        public Builder withUpdateDataOnStart(boolean updateDataOnStart) {

+            this.updateDataOnStart = updateDataOnStart;

+            return this;

+        }

+

+        public LoadIntervalDataOptions build() {

+            return new LoadIntervalDataOptions(name, intervalMs, failOnException, updateDataOnStart);

+        }

+    }

+

+}

diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/LoadIntervalDataUtil.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/LoadIntervalDataUtil.java
new file mode 100644
index 0000000..566d217
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/LoadIntervalDataUtil.java
@@ -0,0 +1,86 @@
+package com.geedgenetworks.core.utils;

+

+import org.apache.flink.shaded.guava18.com.google.common.util.concurrent.ThreadFactoryBuilder;

+import org.apache.flink.util.function.SupplierWithException;

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+import java.util.concurrent.ScheduledExecutorService;

+import java.util.concurrent.ScheduledThreadPoolExecutor;

+import java.util.concurrent.ThreadFactory;

+import java.util.concurrent.TimeUnit;

+import java.util.concurrent.atomic.AtomicBoolean;

+

+public class LoadIntervalDataUtil<T> {

+    static final Logger LOG = LoggerFactory.getLogger(LoadIntervalDataUtil.class);

+

+    private final SupplierWithException<T, Exception> dataSupplier;

+    private final LoadIntervalDataOptions options;

+

+    private final AtomicBoolean started = new AtomicBoolean(false);

+    private final AtomicBoolean stopped = new AtomicBoolean(false);

+    private ScheduledExecutorService scheduler;

+    private volatile Exception exception;

+    private volatile T data;

+

+    private LoadIntervalDataUtil(SupplierWithException<T, Exception> dataSupplier, LoadIntervalDataOptions options) {

+        this.dataSupplier = dataSupplier;

+        this.options = options;

+    }

+

+    public static <T> LoadIntervalDataUtil<T> newInstance(SupplierWithException<T, Exception> dataSupplier, LoadIntervalDataOptions options) {

+        LoadIntervalDataUtil<T> loadIntervalDataUtil = new LoadIntervalDataUtil(dataSupplier, options);

+        loadIntervalDataUtil.start();

+        return loadIntervalDataUtil;

+    }

+

+    public T data() throws Exception {

+        if (!options.failOnException || exception == null) {

+            return data;

+        } else {

+            throw exception;

+        }

+    }

+

+    private void updateData() {

+        try {

+            LOG.info("{} updateData start....", options.name);

+            data = dataSupplier.get();

+            LOG.info("{} updateData end....", options.name);

+        } catch (Throwable t) {

+            if (options.failOnException) {

+                exception = new RuntimeException(t);

+            }

+            LOG.info("{} updateData error", options.name, t);

+        }

+    }

+

+    private void start() {

+        if (started.compareAndSet(false, true)) {

+            if (options.updateDataOnStart) {

+                updateData();

+            }

+            this.scheduler = newDaemonSingleThreadScheduledExecutor(String.format("LoadIntervalDataUtil[%s]", options.name));

+            this.scheduler.scheduleWithFixedDelay(() -> updateData(), options.intervalMs, options.intervalMs, TimeUnit.MILLISECONDS);

+            LOG.info("{} start....", options.name);

+        }

+    }

+

+    public void stop() {

+        if (stopped.compareAndSet(false, true)) {

+            if (scheduler != null) {

+                this.scheduler.shutdown();

+            }

+            LOG.info("{} stop....", options.name);

+        }

+    }

+

+    private static ScheduledExecutorService newDaemonSingleThreadScheduledExecutor(String threadName) {

+        ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat(threadName).build();

+        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, threadFactory);

+        // By default, a cancelled task is not automatically removed from the work queue until its delay

+        // elapses. We have to enable it manually.

+        executor.setRemoveOnCancelPolicy(true);

+        return executor;

+    }

+}

diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/SingleValueMap.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/SingleValueMap.java
new file mode 100644
index 0000000..f6f73c3
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/SingleValueMap.java
@@ -0,0 +1,125 @@
+package com.geedgenetworks.core.utils;

+

+import org.apache.flink.util.function.SupplierWithException;

+import org.slf4j.Logger;

+import org.slf4j.LoggerFactory;

+

+import java.util.Iterator;

+import java.util.LinkedHashMap;

+import java.util.Map;

+import java.util.function.Consumer;

+

+/**

+ * 主要用于实现全局对象

+ * 主要用于flink算子中，方便复用全局对象

+ * flink算子中使用方法:

+ *    // open方法中根据传入的获取data函数获取data

+ *    data = SingleValueMap.acquireData("key", () -> getDataFunc());

+ *    // process方法中使用data

+ *    data.getData()

+ *    // close方法中释放data

+ *    if(data != null)

+ *        data.release();

+ */

+public class SingleValueMap {

+    static final Logger LOG = LoggerFactory.getLogger(SingleValueMap.class);

+    private static Map<Object, Data<?>> cache = new LinkedHashMap<>();

+

+    public static synchronized <T> Data<T> acquireData(Object key, SupplierWithException<T, Exception> dataSupplier) throws Exception {

+        return acquireData(key, dataSupplier, x -> {});

+    }

+

+    public static synchronized <T> Data<T> acquireData(Object key, SupplierWithException<T, Exception> dataSupplier, Consumer<T> releaseFunc) throws Exception {

+        assert releaseFunc != null;

+        Data<?> existingData = cache.get(key);

+        Data<T> data;

+        if (existingData == null) {

+            Data<T> newData = new Data<>(key, dataSupplier.get(), releaseFunc);

+            cache.put(key, newData);

+            data = newData;

+        } else {

+            data = (Data<T>) existingData;

+        }

+        data.useCnt += 1;

+

+        LOG.info("acquireData: {}", data);

+

+        return data;

+    }

+

+    private static synchronized <T> void releaseData(Data<T> data) {

+        Data<?> cachedData = cache.get(data.key);

+        if (cachedData == null) {

+            LOG.error("can not get data: {}", data);

+            return;

+        }

+

+        assert data == cachedData;

+        LOG.info("releaseData: {}", data);

+

+        data.useCnt -= 1;

+        if (!data.inUse()) {

+            data.destroy();

+            cache.remove(data.key);

+

+            LOG.info("removeData: {}", data);

+        }

+    }

+

+    public static synchronized void clear() {

+        Iterator<Map.Entry<Object, Data<?>>> iter = cache.entrySet().iterator();

+        while (iter.hasNext()) {

+            Data<?> data = iter.next().getValue();

+            data.destroy();

+            iter.remove();

+        }

+    }

+

+    public final static class Data<T> {

+        final Object key;

+        final T data;

+        final Consumer<T> destroyFunc;

+        volatile int useCnt = 0;

+

+        Data(Object key, T data, Consumer<T> destroyFunc) {

+            this.key = key;

+            this.data = data;

+            this.destroyFunc = destroyFunc;

+        }

+

+        boolean inUse() {

+            return useCnt > 0;

+        }

+

+        void destroy() {

+            if (destroyFunc != null) {

+                try {

+                    destroyFunc.accept(data);

+                } catch (Exception e) {

+                    LOG.error("error when destroy data: {}", data);

+                }

+            }

+        }

+

+        public void release() {

+            releaseData(this);

+        }

+

+        public Object getKey() {

+            return key;

+        }

+

+        public T getData() {

+            return data;

+        }

+

+        @Override

+        public String toString() {

+            return "Data{" +

+                    "key=" + key +

+                    ", data=" + data +

+                    ", useCnt=" + useCnt +

+                    '}';

+        }

+    }

+}

diff --git a/groot-core/src/test/java/com/geedgenetworks/core/types/TypesTest.java b/groot-core/src/test/java/com/geedgenetworks/core/types/TypesTest.java
index ca8d4e5..518a3f4 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/types/TypesTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/types/TypesTest.java
@@ -1,8 +1,11 @@
 package com.geedgenetworks.core.types;
 
+import com.alibaba.fastjson2.JSON;
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -29,6 +32,42 @@ public class TypesTest {
     }
 
     @Test
+    void test() {
+        Map<String, Object> map = new LinkedHashMap<>();
+        map.put("a", 1);
+        map.put("b", "aa");
+        map.put("c", List.of(1, 2, 3));
+        map.put("int_array", new int[]{1, 2, 3});
+        map.put("str_array", new String[]{"1", "2", "3"});
+        map.put("obj_array", new Object[]{"1", "2", "3"});
+        String jsonString = JSON.toJSONString(map);
+        System.out.println(jsonString);
+    }
+
+    @Test
+    void test2() {
+        Object obj = new int[]{1, 2, 3};
+        System.out.println(obj instanceof byte[]);
+        System.out.println(obj instanceof int[]);
+        System.out.println(obj instanceof String[]);
+        System.out.println(obj instanceof Object[]);
+        System.out.println();
+
+        obj = new String[]{"1", "2", "3"};
+        System.out.println(obj instanceof byte[]);
+        System.out.println(obj instanceof int[]);
+        System.out.println(obj instanceof String[]);
+        System.out.println(obj instanceof Object[]);
+        System.out.println();
+
+        obj = new Object[]{"1", "2", "3"};
+        System.out.println(obj instanceof byte[]);
+        System.out.println(obj instanceof int[]);
+        System.out.println(obj instanceof String[]);
+        System.out.println(obj instanceof Object[]);
+    }
+
+    @Test
     void testParserBaseType() {
         assertEquals(new IntegerType(), Types.parseDataType("INT"));
         assertEquals(new LongType(), Types.parseDataType("biGint"));
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UuidFunctionTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java
index 65e5a94..ef79d51 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UuidFunctionTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java
@@ -3,17 +3,18 @@ package com.geedgenetworks.core.udf.test.simple;
 import com.geedgenetworks.common.Event;
 import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
 import com.geedgenetworks.common.udf.UDFContext;
-import com.geedgenetworks.core.udf.uuid.Uuid;
-import com.geedgenetworks.core.udf.uuid.UuidV5;
-import com.geedgenetworks.core.udf.uuid.UuidV7;
+import com.geedgenetworks.core.udf.uuid.UUID;
+import com.geedgenetworks.core.udf.uuid.UUIDv5;
+import com.geedgenetworks.core.udf.uuid.UUIDv7;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.util.*;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 
-public class UuidFunctionTest {
+public class UUIDTest {
 
     private UDFContext udfContext;
     private Map<String, Object> parameters ;
@@ -22,7 +23,7 @@ public class UuidFunctionTest {
     @Test
     public void testInit(){
         udfContext = new UDFContext();
-        UuidV5 uuidv5 = new UuidV5();
+        UUIDv5 uuidv5 = new UUIDv5();
         parameters =  new HashMap<>();
         udfContext.setParameters(parameters);
         udfContext.setLookup_fields(List.of("client_ip","server_ip"));
@@ -35,9 +36,9 @@ public class UuidFunctionTest {
     }
 
     @Test
-    public void testUuid() {
+    public void testUUID() {
         udfContext = new UDFContext();
-        Uuid uuid = new Uuid();
+        UUID uuid = new UUID();
         parameters =  new HashMap<>();
         udfContext.setParameters(parameters);
         udfContext.setOutput_fields(Collections.singletonList("uuid"));
@@ -49,9 +50,9 @@ public class UuidFunctionTest {
         assertEquals(36, result1.getExtractedFields().get("uuid").toString().length());
     }
     @Test
-    public void testUuidV7() {
+    public void testUUIDV7() {
         udfContext = new UDFContext();
-        UuidV7 uuid = new UuidV7();
+        UUIDv7 uuid = new UUIDv7();
         parameters =  new HashMap<>();
         udfContext.setParameters(parameters);
         udfContext.setOutput_fields(Collections.singletonList("uuid"));
@@ -63,28 +64,30 @@ public class UuidFunctionTest {
         assertEquals(36, result1.getExtractedFields().get("uuid").toString().length());
     }
     @Test
-    public void testUuidV5ForNameSpaceIp() {
+    public void testUUIDV5ForNameSpaceIp() {
         udfContext = new UDFContext();
-        UuidV5 uuidv5 = new UuidV5();
+        UUIDv5 uuidv5 = new UUIDv5();
         parameters =  new HashMap<>();
         udfContext.setParameters(parameters);
-        udfContext.setLookup_fields(List.of("client_ip","server_ip"));
+        udfContext.setLookup_fields(List.of("client_ip", "server_ip"));
         udfContext.setOutput_fields(Collections.singletonList("uuid"));
         parameters.put("namespace","NAMESPACE_IP");
         uuidv5.open(null, udfContext);
         Event event = new Event();
         Map<String, Object> extractedFields = new HashMap<>();
         extractedFields.put("client_ip", "1.1.1.1");
-        extractedFields.put("server_ip", "1.1.1.2");
+        extractedFields.put("server_ip", "");
         event.setExtractedFields(extractedFields);
-        Event result1 = uuidv5.evaluate(event);
-        assertEquals("52530d0c-07df-5c4b-a659-661242575386", result1.getExtractedFields().get("uuid").toString());
+        Event result = uuidv5.evaluate(event);
+        System.out.printf("uuid: %s\n", result.getExtractedFields().get("uuid").toString());
+        assertEquals("5394a6a8-b9b8-5147-b5b2-01365f158acb", result.getExtractedFields().get("uuid").toString());
+        assertNotEquals("ecc67867-1f76-580c-a4c1-6a3d16ad6d02", result.getExtractedFields().get("uuid").toString());
     }
 
     @Test
-    public void testUuidV5ForNameSpaceDomain() {
+    public void testUUIDV5ForNameSpaceDomain() {
         udfContext = new UDFContext();
-        UuidV5 uuidv5 = new UuidV5();
+        UUIDv5 uuidv5 = new UUIDv5();
         parameters =  new HashMap<>();
         udfContext.setParameters(parameters);
         udfContext.setLookup_fields(List.of("domain"));
@@ -99,9 +102,9 @@ public class UuidFunctionTest {
         assertEquals("fd67cec1-6b33-5def-835c-fbe32f1ce4a4", result1.getExtractedFields().get("uuid").toString());
     }
     @Test
-    public void testUuidV5ForNameSpaceApp() {
+    public void testUUIDv5ForNameSpaceApp() {
         udfContext = new UDFContext();
-        UuidV5 uuidv5 = new UuidV5();
+        UUIDv5 uuidv5 = new UUIDv5();
         parameters =  new HashMap<>();
         udfContext.setParameters(parameters);
         udfContext.setLookup_fields(List.of("app"));
@@ -117,18 +120,18 @@ public class UuidFunctionTest {
     }
 
     @Test
-    public void testUuidV5ForNameSpaceSubid() {
+    public void testUUIDV5ForNameSpaceSubscriberID() {
         udfContext = new UDFContext();
-        UuidV5 uuidv5 = new UuidV5();
+        UUIDv5 uuidv5 = new UUIDv5();
         parameters =  new HashMap<>();
         udfContext.setParameters(parameters);
-        udfContext.setLookup_fields(List.of("subid"));
+        udfContext.setLookup_fields(List.of("subscriber_id"));
         udfContext.setOutput_fields(Collections.singletonList("uuid"));
         parameters.put("namespace","NAMESPACE_SUBSCRIBER");
         uuidv5.open(null, udfContext);
         Event event = new Event();
         Map<String, Object> extractedFields = new HashMap<>();
-        extractedFields.put("subid", "test1");
+        extractedFields.put("subscriber_id", "test1");
         event.setExtractedFields(extractedFields);
         Event result1 = uuidv5.evaluate(event);
         assertEquals("9b154520-3c29-541c-bb81-f649354dae67", result1.getExtractedFields().get("uuid").toString());
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/utils/LoadIntervalDataUtilTest.java b/groot-core/src/test/java/com/geedgenetworks/core/utils/LoadIntervalDataUtilTest.java
new file mode 100644
index 0000000..b7c6306
--- /dev/null
+++ b/groot-core/src/test/java/com/geedgenetworks/core/utils/LoadIntervalDataUtilTest.java
@@ -0,0 +1,80 @@
+package com.geedgenetworks.core.utils;

+

+

+import java.sql.Timestamp;

+

+public class LoadIntervalDataUtilTest {

+

+    public static void main(String[] args)  throws Exception{

+        //testNoError();

+        //testNotUpdateDataOnStart();

+        //testWithErrorAndNotFail();

+        testWithErrorAndFail();

+    }

+

+    public static void testNoError()  throws Exception{

+        LoadIntervalDataUtil<Timestamp> util = LoadIntervalDataUtil.newInstance(() -> new Timestamp(System.currentTimeMillis()),

+                LoadIntervalDataOptions.defaults("time", 3000));

+

+        System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+

+        for (int i = 0; i < 10; i++) {

+            Thread.sleep(1000);

+            System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+        }

+

+        util.stop();

+    }

+

+    public static void testNotUpdateDataOnStart()  throws Exception{

+        LoadIntervalDataUtil<Timestamp> util = LoadIntervalDataUtil.newInstance(() -> new Timestamp(System.currentTimeMillis()),

+                LoadIntervalDataOptions.builder().withName("time").withIntervalMs(3000).withUpdateDataOnStart(false).build());

+

+        System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+

+        for (int i = 0; i < 10; i++) {

+            Thread.sleep(1000);

+            System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+        }

+

+        util.stop();

+    }

+

+    public static void testWithErrorAndNotFail() throws Exception{

+        final long start = System.currentTimeMillis();

+        LoadIntervalDataUtil<Timestamp> util = LoadIntervalDataUtil.newInstance(() -> {

+            if(System.currentTimeMillis() - start >= 5000){

+                throw new RuntimeException(new Timestamp(System.currentTimeMillis()).toString());

+            }

+            return new Timestamp(System.currentTimeMillis());

+         }, LoadIntervalDataOptions.defaults("time", 3000));

+

+        System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+

+        for (int i = 0; i < 10; i++) {

+            Thread.sleep(1000);

+            System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+        }

+

+        util.stop();

+    }

+

+    public static void testWithErrorAndFail() throws Exception{

+        final long start = System.currentTimeMillis();

+        LoadIntervalDataUtil<Timestamp> util = LoadIntervalDataUtil.newInstance(() -> {

+            if(System.currentTimeMillis() - start >= 5000){

+                throw new RuntimeException(new Timestamp(System.currentTimeMillis()).toString());

+            }

+            return new Timestamp(System.currentTimeMillis());

+        }, LoadIntervalDataOptions.builder().withName("time").withIntervalMs(3000).withFailOnException(true).build());

+

+        System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+

+        for (int i = 0; i < 10; i++) {

+            Thread.sleep(1000);

+            System.out.println(new Timestamp(System.currentTimeMillis()) + " - " + util.data());

+        }

+

+        util.stop();

+    }

+}
\ No newline at end of file
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/utils/SingleValueMapTest.java b/groot-core/src/test/java/com/geedgenetworks/core/utils/SingleValueMapTest.java
new file mode 100644
index 0000000..f5f1e7c
--- /dev/null
+++ b/groot-core/src/test/java/com/geedgenetworks/core/utils/SingleValueMapTest.java
@@ -0,0 +1,98 @@
+package com.geedgenetworks.core.utils;

+

+import org.junit.jupiter.api.Assertions;

+

+import java.sql.Timestamp;

+import java.util.concurrent.ThreadLocalRandom;

+import java.util.concurrent.atomic.AtomicInteger;

+

+public class SingleValueMapTest {

+

+    public static void main(String[] args) throws Exception {

+        //testSingleValue();

+        testSingleValueWithLoadIntervalDataUtil();

+    }

+

+    public static void testSingleValue() throws Exception {

+        Thread[] threads = new Thread[20];

+        for (int i = 0; i < threads.length; i++) {

+            threads[i] = new Thread(() -> {

+                SingleValueMap.Data<ConnDada> connDada = null;

+                try {

+                    connDada = SingleValueMap.acquireData("conn_data", () -> new ConnDada(), x -> {

+                        System.out.println("close conn");

+                    });

+                } catch (Exception e) {

+                    throw new RuntimeException(e);

+                }

+

+                try {

+                    Thread.sleep(ThreadLocalRandom.current().nextInt(5) * 10);

+                } catch (InterruptedException e) {

+                    throw new RuntimeException(e);

+                }

+

+                connDada.release();

+            }, "Thread-" + i);

+        }

+

+        for (int i = 0; i < threads.length; i++) {

+            threads[i].start();

+        }

+

+        for (int i = 0; i < threads.length; i++) {

+            threads[i].join();

+        }

+

+        System.out.println("initCnt:" + ConnDada.initCnt.get());

+        Assertions.assertEquals(ConnDada.initCnt.get(), 1);

+    }

+

+    public static void testSingleValueWithLoadIntervalDataUtil() throws Exception {

+        Thread[] threads = new Thread[20];

+        for (int i = 0; i < threads.length; i++) {

+            threads[i] = new Thread(() -> {

+                SingleValueMap.Data<LoadIntervalDataUtil<Timestamp>> util = null;

+                try {

+                    util = SingleValueMap.acquireData("LoadIntervalDataUtil",

+                            () ->  LoadIntervalDataUtil.newInstance(() -> new Timestamp(System.currentTimeMillis()), LoadIntervalDataOptions.defaults("time", 3000)),

+                            LoadIntervalDataUtil::stop);

+                } catch (Exception e) {

+                    throw new RuntimeException(e);

+                }

+

+

+                try {

+                    for (int j = 0; j < 10; j++) {

+                        Thread.sleep(1000);

+                        System.out.println(Thread.currentThread().getName() + " - " + new Timestamp(System.currentTimeMillis()) + " - " + util.getData().data());

+                    }

+

+                    Thread.sleep(ThreadLocalRandom.current().nextInt(5) * 10);

+                } catch (Exception e) {

+                    throw new RuntimeException(e);

+                }

+

+                util.release();

+            }, "Thread-" + i);

+        }

+

+        for (int i = 0; i < threads.length; i++) {

+            threads[i].start();

+        }

+

+        for (int i = 0; i < threads.length; i++) {

+            threads[i].join();

+        }

+

+    }

+

+    public static class ConnDada {

+        static AtomicInteger initCnt = new AtomicInteger(0);

+        public ConnDada(){

+            System.out.println("ConnDada init");

+            initCnt.incrementAndGet();

+        }

+

+    }

+}
\ No newline at end of file