Merge branch 'improve/uuidv5' into 'release/1.7.0'release/1.7.0

[Improve][Encrypt] Enhance Encrypt is applied to encryption at transit and...

See merge request galaxy/platform/groot-stream!134

20 files changed, 753 insertions, 322 deletions

diff --git a/groot-core/src/main/java/com/geedgenetworks/core/pojo/DataEncryptionKey.java b/groot-core/src/main/java/com/geedgenetworks/core/pojo/DataEncryptionKey.java
new file mode 100644
index 0000000..8c0d516
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/pojo/DataEncryptionKey.java
@@ -0,0 +1,19 @@
+package com.geedgenetworks.core.pojo;
+
+
+import lombok.Data;
+
+@Data
+public class DataEncryptionKey {
+
+    private byte[] data;
+    private int version;
+
+    public DataEncryptionKey() {
+    }
+
+    public DataEncryptionKey(byte[] data, int version) {
+        this.data = data;
+        this.version = version;
+    }
+}
\ No newline at end of file
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/pojo/KmsKey.java b/groot-core/src/main/java/com/geedgenetworks/core/pojo/KmsKey.java
deleted file mode 100644
index 2690254..0000000
--- a/groot-core/src/main/java/com/geedgenetworks/core/pojo/KmsKey.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package com.geedgenetworks.core.pojo;
-
-
-import lombok.Data;
-
-@Data
-public class KmsKey {
-
-    private byte[] keyData;
-    private int keyVersion;
-
-    public KmsKey() {
-    }
-
-    public KmsKey(byte[] keyData, int keyVersion) {
-        this.keyData = keyData;
-        this.keyVersion = keyVersion;
-    }
-}
\ No newline at end of file
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java
index 74816f5..9046472 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Domain.java
@@ -100,10 +100,9 @@ public class Domain implements ScalarFunction {
             throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg());
         }
 
-        if(!udfContext.getParameters().containsKey(UDFContextConfigOptions.PARAMETERS_OPTION.key())){
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format(
-                    "UDF: %s, [%s] Option should be specified.",
-                    udfContext.getFunction(), UDFContextConfigOptions.PARAMETERS.key()));
+        result = CheckUDFContextUtil.checkParametersContainsKeys(udfContext, UDFContextConfigOptions.PARAMETERS_OPTION.key());
+        if (!result.isSuccess()) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg());
         }
 
         String optionValue = udfContext.getParameters().get(UDFContextConfigOptions.PARAMETERS_OPTION.key()).toString();
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java
index c8e21b2..2fa0804 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Encrypt.java
@@ -1,35 +1,28 @@
 package com.geedgenetworks.core.udf;
 
 import cn.hutool.core.util.URLUtil;
-import cn.hutool.json.JSONArray;
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
 import com.alibaba.fastjson2.JSON;
 import com.geedgenetworks.common.Constants;
 import com.geedgenetworks.common.Event;
-import com.geedgenetworks.common.config.CommonConfig;
-import com.geedgenetworks.common.config.KmsConfig;
-import com.geedgenetworks.common.config.SSLConfig;
+import com.geedgenetworks.common.config.*;
 import com.geedgenetworks.common.exception.CommonErrorCode;
 import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
 import com.geedgenetworks.common.udf.ScalarFunction;
 import com.geedgenetworks.common.udf.UDFContext;
-import com.geedgenetworks.core.pojo.KmsKey;
-import com.geedgenetworks.core.udf.encrypt.EncryptionAlgorithm;
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
+import com.geedgenetworks.core.udf.encrypt.Crypto;
 import com.geedgenetworks.core.utils.*;
 import com.geedgenetworks.shaded.org.apache.http.HttpHeaders;
 import com.geedgenetworks.shaded.org.apache.http.HttpStatus;
 import com.geedgenetworks.shaded.org.apache.http.message.BasicHeader;
-import com.geedgenetworks.utils.StringUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.configuration.Configuration;
 
-import java.io.IOException;
 import java.net.URI;
-import java.util.Arrays;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -40,86 +33,122 @@ public class Encrypt implements ScalarFunction {
     private String outputFieldName;
     private String identifier;
     private String defaultVal;
-    private String type;
-    private transient SingleValueMap.Data<LoadIntervalDataUtil<Set<String>>> sensitiveFieldsData;
-    private transient SingleValueMap.Data<LoadIntervalDataUtil<KmsKey>> kmsKeyData;
-    private transient EncryptionAlgorithm encryptionAlgorithm;
+    private String kmsType;
+    private transient SingleValueMap.Data<LoadIntervalDataUtil<Set<String>>> sensitiveFieldState;
+    private transient SingleValueMap.Data<LoadIntervalDataUtil<DataEncryptionKey>> dekState;
+    private transient Crypto crypto;
 
     @Override
     public void open(RuntimeContext runtimeContext, UDFContext udfContext) {
-        checkUdfContext(udfContext);
-        if (udfContext.getParameters().containsKey("default_val")) {
-            this.defaultVal = udfContext.getParameters().get("default_val").toString();
-        }
+        checkConfig(udfContext);
+        configureParameters(udfContext);
+        initializeCrypto(runtimeContext);
+    }
+
+    private void configureParameters(UDFContext udfContext) {
+        this.defaultVal = Optional.ofNullable(udfContext.getParameters().get("default_val"))
+                .map(Object::toString).orElse("");
         this.lookupFieldName = udfContext.getLookupFields().get(0);
         this.outputFieldName = udfContext.getOutputFields().get(0);
-        this.identifier = udfContext.getParameters().get("identifier").toString();
+        this.identifier = udfContext.getParameters().get(UDFContextConfigOptions.PARAMETERS_IDENTIFIER.key()).toString();
+    }
+
+    private void initializeCrypto(RuntimeContext runtimeContext) {
         Configuration configuration = (Configuration) runtimeContext.getExecutionConfig().getGlobalJobParameters();
         CommonConfig commonConfig = JSON.parseObject(configuration.toMap().get(Constants.SYSPROP_GROOTSTREAM_CONFIG), CommonConfig.class);
         KmsConfig kmsConfig = commonConfig.getKmsConfig().get(configuration.toMap().get(Constants.SYSPROP_KMS_TYPE_CONFIG));
-        SSLConfig sslConfig = commonConfig.getSslConfig();
+        ClientSSLConfig sslConfig = commonConfig.getSslConfig();
         Map<String, String> propertiesConfig = commonConfig.getPropertiesConfig();
-        type = kmsConfig.getType();
+        this.kmsType = kmsConfig.getType();
         try {
-            encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(identifier);
-            if (encryptionAlgorithm == null) {
-                throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Parameters identifier is illegal!");
+            this.crypto = CryptoProvider.createEncryptionAlgorithm(identifier);
+            initializeDataEncryptionKeyIfNeeded(kmsConfig, sslConfig, propertiesConfig);
+            initializeSensitiveFieldsIfNeeded(propertiesConfig);
+        } catch (Exception e) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Initialization UDF Encrypt failed!", e);
+        }
+    }
+
+    private void initializeDataEncryptionKeyIfNeeded(KmsConfig kmsConfig, ClientSSLConfig sslConfig, Map<String, String> propertiesConfig) throws Exception {
+        if (KMSUtils.KMS_TYPE_VAULT.equals(kmsType)) {
+            this.dekState = SingleValueMap.acquireData("dekState",
+                    () -> LoadIntervalDataUtil.newInstance(
+                            () -> KMSUtils.getVaultKey(kmsConfig, sslConfig, identifier),
+                            LoadIntervalDataOptions.defaults("dekState",
+                                    Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "1")) * 60000L)),
+                    LoadIntervalDataUtil::stop);
+
+            DataEncryptionKey dek = dekState.getData().data();
+
+            if (dek == null) {
+                throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format("URI: %s, Unable to get KMS Secret Key!", kmsConfig.getUrl()));
             }
-            if (!type.equals(KmsUtils.KMS_TYPE_LOCAL)) {
-                kmsKeyData = SingleValueMap.acquireData("kmsKeyData",
-                        () -> LoadIntervalDataUtil.newInstance(() -> KmsUtils.getVaultKey(kmsConfig, sslConfig, identifier),
-                                LoadIntervalDataOptions.defaults("kmsKeyData", Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "5")) * 60000L)),
-                        LoadIntervalDataUtil::stop);
-                KmsKey kmsKey = kmsKeyData.getData().data();
-                if (kmsKey == null) {
-                    throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Initialization UDF Encrypt failed!");
-                }
-                if (encryptionAlgorithm.getSecretKeyLength() != kmsKey.getKeyData().length) {
-                    throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Global parameter kms secret Key requires " + encryptionAlgorithm.getSecretKeyLength() + " bytes!");
-                }
-                encryptionAlgorithm.setKmsKey(kmsKey);
+
+            if (crypto.getSecretKeyLength() != dek.getData().length) {
+                throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format("Version: %s, [%s] KMS Secret Key requires %s bytes!",
+                        dek.getVersion(), crypto.getIdentifier(), crypto.getSecretKeyLength()));
             }
-            sensitiveFieldsData = SingleValueMap.acquireData("sensitiveFields",
-                    () -> LoadIntervalDataUtil.newInstance(() -> getSensitiveFields(propertiesConfig.get("projection.encrypt.schema.registry.uri")),
-                            LoadIntervalDataOptions.defaults("sensitiveFields", Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "5")) * 60000L)),
+
+            crypto.setDataEncryptionKey(dek);
+        }
+    }
+
+    private void initializeSensitiveFieldsIfNeeded(Map<String, String> propertiesConfig) throws Exception {
+        String schemaUri = propertiesConfig.get("projection.encrypt.schema.registry.uri");
+        if (schemaUri != null) {
+            this.sensitiveFieldState = SingleValueMap.acquireData("sensitiveFieldState",
+                    () -> LoadIntervalDataUtil.newInstance(
+                            () -> getSensitiveFields(schemaUri),
+                            LoadIntervalDataOptions.defaults("sensitiveFieldState",
+                                    Integer.parseInt(propertiesConfig.getOrDefault(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "1")) * 60000L)),
                     LoadIntervalDataUtil::stop);
-            if (sensitiveFieldsData.getData().data() == null) {
-                throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Initialization UDF Encrypt failed!");
+            if (sensitiveFieldState.getData().data() == null) {
+                throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, String.format("URI: %s, Unable to get encrypted fields!", schemaUri));
             }
-        } catch (Exception e) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Initialization UDF Encrypt failed!", e);
         }
     }
 
+
     @Override
     public Event evaluate(Event event) {
         try {
-            if (!type.equals(KmsUtils.KMS_TYPE_LOCAL)) {
-                KmsKey kmsKey = kmsKeyData.getData().data();
-                if (kmsKey.getKeyVersion() != encryptionAlgorithm.getKmsKey().getKeyVersion() || !Arrays.equals(kmsKey.getKeyData(), encryptionAlgorithm.getKmsKey().getKeyData())) {
-                    encryptionAlgorithm.setKmsKey(kmsKey);
-                }
-            }
-            if (sensitiveFieldsData.getData().data().contains(lookupFieldName) && event.getExtractedFields().containsKey(lookupFieldName)) {
-                String value = (String) event.getExtractedFields().get(lookupFieldName);
-                if (StringUtil.isNotBlank(value)) {
-                    String encryptResult = encryptionAlgorithm.encrypt(value);
-                    if (StringUtil.isEmpty(encryptResult)) {
-                        event.getExtractedFields().put(outputFieldName, StringUtil.isNotBlank(defaultVal) ? defaultVal : value);
-                    } else {
-                        if (KmsUtils.KMS_TYPE_VAULT.equals(type)) {
-                            encryptResult = "vault:v" + encryptionAlgorithm.getKmsKey().getKeyVersion() + ":" + encryptResult;
-                        }
-                        event.getExtractedFields().put(outputFieldName, encryptResult);
-                    }
-                }
+            refreshDataEncryptionKey();
+
+            if (isSensitiveField(lookupFieldName)) {
+                executeEncrypt(event);
             }
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            log.error("Encrypt failed!", e);
         }
         return event;
     }
 
+    private void refreshDataEncryptionKey() throws Exception {
+        if (dekState == null || dekState.getData().data() == null) {
+            return;
+        }
+        DataEncryptionKey dek = dekState.getData().data();
+        if (!Arrays.equals(dek.getData(), crypto.getDataEncryptionKey().getData())) {
+            crypto.setDataEncryptionKey(dek);
+        }
+
+    }
+
+    private boolean isSensitiveField(String fieldName) throws Exception {
+        return sensitiveFieldState == null || sensitiveFieldState.getData().data().contains(fieldName);
+    }
+
+    private void executeEncrypt(Event event) {
+        Object value = event.getExtractedFields().get(lookupFieldName);
+        if (value != null) {
+            String encryptResult = Optional.ofNullable(crypto.encrypt(value.toString())).orElse(defaultVal);
+            if (KMSUtils.KMS_TYPE_VAULT.equals(kmsType)) {
+                encryptResult = "vault:v" + crypto.getDataEncryptionKey().getVersion() + ":" + encryptResult;
+            }
+            event.getExtractedFields().put(outputFieldName, encryptResult);
+        }
+    }
+
     @Override
     public String functionName() {
         return "ENCRYPT";
@@ -127,41 +156,65 @@ public class Encrypt implements ScalarFunction {
 
     @Override
     public void close() {
-        if (sensitiveFieldsData != null) {
-            sensitiveFieldsData.release();
+        if (sensitiveFieldState != null) {
+            sensitiveFieldState.release();
         }
-        if (kmsKeyData != null) {
-            kmsKeyData.release();
+        if (dekState != null) {
+            dekState.release();
         }
     }
 
-    private void checkUdfContext(UDFContext udfContext) {
-        if (udfContext.getParameters() == null) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Missing required parameters");
+    @Override
+    public void checkConfig(UDFContext udfContext) {
+
+        CheckResult result = CheckUDFContextUtil.checkAllExists(udfContext,
+                UDFContextConfigOptions.LOOKUP_FIELDS.key(),
+                UDFContextConfigOptions.OUTPUT_FIELDS.key(),
+                UDFContextConfigOptions.PARAMETERS.key());
+
+        if (!result.isSuccess()) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.CONFIG_VALIDATION_FAILED, result.getMsg());
         }
-        if (udfContext.getLookupFields().size() != 1) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function lookup fields only support 1 value");
+
+        result = CheckUDFContextUtil.checkCollectionSingleItemExists(udfContext, UDFContextConfigOptions.LOOKUP_FIELDS.key());
+        if (!result.isSuccess()) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg());
         }
-        if (udfContext.getOutputFields().size() != 1) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function output fields only support 1 value");
+
+        result = CheckUDFContextUtil.checkCollectionSingleItemExists(udfContext, UDFContextConfigOptions.OUTPUT_FIELDS.key());
+        if (!result.isSuccess()) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg());
         }
-        if (!udfContext.getParameters().containsKey("identifier")) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Parameters must contains identifier");
+
+        result = CheckUDFContextUtil.checkParametersContainsKeys(udfContext, UDFContextConfigOptions.PARAMETERS_IDENTIFIER.key());
+        if (!result.isSuccess()) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg());
         }
+
+
     }
 
-    public Set<String> getSensitiveFields(String url) throws IOException {
-        Set<String> sensitiveFieldsSet;
-        String sensitiveFieldsStr = HttpClientPoolUtil.getInstance().httpGet(URI.create(URLUtil.normalize(url)), new BasicHeader(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded"));
-        JSONObject sensitiveFieldsJson = JSONUtil.parseObj(sensitiveFieldsStr);
-        if (sensitiveFieldsJson.getInt("status", HttpStatus.SC_INTERNAL_SERVER_ERROR) == HttpStatus.SC_OK) {
-            JSONArray sensitiveFieldsJsonArr = sensitiveFieldsJson.getJSONArray("data");
-            sensitiveFieldsSet = IntStream.range(0, sensitiveFieldsJsonArr.size())
-                    .mapToObj(sensitiveFieldsJsonArr::getStr)
-                    .collect(Collectors.toSet());
-        } else {
-            throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Get encrypt fields error! Error message: " + sensitiveFieldsStr);
+    // If accessing the schema URI fails, the last known state of sensitiveFieldState will be used
+    private Set<String> getSensitiveFields(String url) throws Exception {
+        try {
+            String result = HttpClientPoolUtil.getInstance()
+                    .httpGet(URI.create(URLUtil.normalize(url)), new BasicHeader(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded"));
+            JSONObject resultJson = JSONUtil.parseObj(result);
+            int statusCode = resultJson.getInt("status", HttpStatus.SC_INTERNAL_SERVER_ERROR);
+            if (statusCode == HttpStatus.SC_OK) {
+                return Optional.ofNullable(resultJson.getJSONArray("data"))
+                        .map(jsonArray -> IntStream.range(0, jsonArray.size())
+                                .mapToObj(jsonArray::getStr)
+                                .collect(Collectors.toSet()))
+                        .orElseGet(Collections::emptySet);
+            } else {
+                log.error("Get sensitive fields error! Error message: {}", result);
+                return sensitiveFieldState.getData().data();
+            }
+        } catch (Exception e) {
+            log.error("Get sensitive fields error! Error message: {}", e.getMessage());
+            return sensitiveFieldState.getData().data(); // return the previous value
         }
-        return sensitiveFieldsSet;
     }
+
 }
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java
index 098cdef..a18d361 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/Hmac.java
@@ -3,6 +3,9 @@ package com.geedgenetworks.core.udf;
 import cn.hutool.crypto.digest.HMac;
 import cn.hutool.crypto.digest.HmacAlgorithm;
 import com.geedgenetworks.common.Event;
+import com.geedgenetworks.common.config.CheckResult;
+import com.geedgenetworks.common.config.CheckUDFContextUtil;
+import com.geedgenetworks.common.config.UDFContextConfigOptions;
 import com.geedgenetworks.common.exception.CommonErrorCode;
 import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
 import com.geedgenetworks.common.udf.ScalarFunction;
@@ -21,7 +24,7 @@ public class Hmac implements ScalarFunction {
 
     @Override
     public void open(RuntimeContext runtimeContext, UDFContext udfContext) {
-        checkUdfContext(udfContext);
+        checkConfig(udfContext);
         String secretKey = udfContext.getParameters().get("secret_key").toString();
         String algorithm = "sha256";
         if (udfContext.getParameters().containsKey("algorithm")) {
@@ -39,21 +42,22 @@ public class Hmac implements ScalarFunction {
     @Override
     public Event evaluate(Event event) {
         String encodeResult = "";
-        String message = (String) event.getExtractedFields().get(lookupFieldName);
-        if (StringUtil.isNotBlank(message)) {
+        Object value = event.getExtractedFields().get(lookupFieldName);
+        if (StringUtil.isNotEmpty(value)) {
             switch (outputFormat) {
                 case "hex":
-                    encodeResult = hMac.digestHex(message);
+                    encodeResult = hMac.digestHex(value.toString());
                     break;
                 case "base64":
-                    encodeResult = hMac.digestBase64(message, false);
+                    encodeResult = hMac.digestBase64(value.toString(), false);
                     break;
                 default:
-                    encodeResult = hMac.digestBase64(message, false);
+                    encodeResult = hMac.digestBase64(value.toString(), false);
                     break;
             }
+            event.getExtractedFields().put(outputFieldName, encodeResult);
         }
-        event.getExtractedFields().put(outputFieldName, encodeResult);
+
         return event;
     }
 
@@ -67,21 +71,25 @@ public class Hmac implements ScalarFunction {
 
     }
 
-    private void checkUdfContext(UDFContext udfContext) {
-        if (udfContext.getParameters() == null || udfContext.getOutputFields() == null) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "Missing required parameters");
-        }
-        if (udfContext.getLookupFields().size() != 1) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function lookup fields only support 1 value");
-        }
-        if (udfContext.getOutputFields().size() != 1) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "The function output fields only support 1 value");
+    @Override
+    public void checkConfig(UDFContext udfContext) {
+
+        CheckResult result = CheckUDFContextUtil.checkAllExists(udfContext,
+                UDFContextConfigOptions.LOOKUP_FIELDS.key(),
+                UDFContextConfigOptions.OUTPUT_FIELDS.key(),
+                UDFContextConfigOptions.PARAMETERS.key());
+
+        if (!result.isSuccess()) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.CONFIG_VALIDATION_FAILED, result.getMsg());
         }
-        if (!udfContext.getParameters().containsKey("secret_key")) {
-            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, "parameters must contains secret_key");
+        result = CheckUDFContextUtil.checkParametersContainsKeys(udfContext, UDFContextConfigOptions.PARAMETERS_SECRET_KEY.key());
+        if (!result.isSuccess()) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.ILLEGAL_ARGUMENT, result.getMsg());
         }
+
     }
 
+
     private String getHmacAlgorithm(String algorithm) {
         if (StringUtil.containsIgnoreCase(algorithm, "sha256")) {
             return HmacAlgorithm.HmacSHA256.getValue();
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM.java
new file mode 100644
index 0000000..f08383a
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM.java
@@ -0,0 +1,81 @@
+package com.geedgenetworks.core.udf.encrypt;
+
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
+
+import javax.crypto.Cipher;
+import javax.crypto.spec.GCMParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+import java.util.Base64;
+
+public class AES128GCM implements Crypto {
+    private static final String IDENTIFIER = "aes-128-gcm";
+    private static final String ALGORITHM = "AES";
+    private static final String TRANSFORMATION = "AES/GCM/NoPadding";
+    private static final int GCM_TAG_LENGTH = 128;
+    private static final int SECRET_KEY_LENGTH = 16;
+    private static final byte[] DEFAULT_SECRET_KEY = ".geedgenetworks.".getBytes();
+    private static final byte[] NONCE = "Galaxy2019#*".getBytes();
+
+    private DataEncryptionKey dek;
+
+    public AES128GCM() {
+        this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1);
+    }
+
+    @Override
+    public String getIdentifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public int getSecretKeyLength() {
+        return SECRET_KEY_LENGTH;
+    }
+
+    @Override
+    public DataEncryptionKey getDataEncryptionKey() {
+        return dek;
+    }
+
+    @Override
+    public void setDataEncryptionKey(DataEncryptionKey dek) {
+        this.dek = dek;
+    }
+
+    @Override
+    public String encrypt(String content) {
+        String encryptedString = "";
+        try {
+            Cipher cipher = Cipher.getInstance(TRANSFORMATION);
+            GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE);
+            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
+            byte[] encryptedBytes = cipher.doFinal(content.getBytes());
+            byte[] combinedBytes = new byte[NONCE.length + encryptedBytes.length];
+            System.arraycopy(NONCE, 0, combinedBytes, 0, NONCE.length);
+            System.arraycopy(encryptedBytes, 0, combinedBytes, NONCE.length, encryptedBytes.length);
+            encryptedString = Base64.getEncoder().encodeToString(combinedBytes);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+        return encryptedString;
+    }
+
+    @Override
+    public String decrypt(String content) {
+        String decryptedString = "";
+        try {
+            byte[] combined = Base64.getDecoder().decode(content);
+            byte[] encryptedBytes = new byte[combined.length - NONCE.length];
+            System.arraycopy(combined, 0, NONCE, 0, NONCE.length);
+            System.arraycopy(combined, NONCE.length, encryptedBytes, 0, encryptedBytes.length);
+            GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, NONCE);
+            Cipher cipher = Cipher.getInstance(TRANSFORMATION);
+            cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
+            byte[] decryptedBytes = cipher.doFinal(encryptedBytes);
+            decryptedString = new String(decryptedBytes);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+        return decryptedString;
+    }
+}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96.java
index 90669b3..6ca1af2 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES128GCM96.java
@@ -1,14 +1,18 @@
 package com.geedgenetworks.core.udf.encrypt;
 
 import cn.hutool.core.util.RandomUtil;
-import com.geedgenetworks.core.pojo.KmsKey;
+import com.geedgenetworks.common.exception.CommonErrorCode;
+import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
 
 import javax.crypto.Cipher;
+import javax.crypto.NoSuchPaddingException;
 import javax.crypto.spec.GCMParameterSpec;
 import javax.crypto.spec.SecretKeySpec;
+import java.security.NoSuchAlgorithmException;
 import java.util.Base64;
 
-public class AES128GCM96 implements EncryptionAlgorithm {
+public class AES128GCM96 implements Crypto {
     private static final String IDENTIFIER = "aes-128-gcm96";
     private static final String ALGORITHM = "AES";
     private static final String TRANSFORMATION = "AES/GCM/NoPadding";
@@ -18,11 +22,15 @@ public class AES128GCM96 implements EncryptionAlgorithm {
     private static final byte[] DEFAULT_SECRET_KEY = ".geedgenetworks.".getBytes();
 
     private final Cipher cipher;
-    private KmsKey kmsKey;
+    private DataEncryptionKey dek;
 
-    public AES128GCM96() throws Exception {
-        this.cipher = Cipher.getInstance(TRANSFORMATION);
-        this.kmsKey = new KmsKey(DEFAULT_SECRET_KEY, 1);
+    public AES128GCM96() {
+        try {
+            this.cipher = Cipher.getInstance(TRANSFORMATION);
+            this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1);
+        } catch (NoSuchAlgorithmException | NoSuchPaddingException e) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Failed to initialize AES128GCM96", e);
+        }
     }
 
     @Override
@@ -36,13 +44,13 @@ public class AES128GCM96 implements EncryptionAlgorithm {
     }
 
     @Override
-    public KmsKey getKmsKey() {
-        return kmsKey;
+    public DataEncryptionKey getDataEncryptionKey() {
+        return dek;
     }
 
     @Override
-    public void setKmsKey(KmsKey kmsKey) {
-        this.kmsKey = kmsKey;
+    public void setDataEncryptionKey(DataEncryptionKey dek) {
+        this.dek = dek;
     }
 
     @Override
@@ -51,7 +59,7 @@ public class AES128GCM96 implements EncryptionAlgorithm {
         try {
             byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH);
             GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce);
-            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec);
+            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
             byte[] encryptedBytes = cipher.doFinal(content.getBytes());
             byte[] combinedBytes = new byte[GCM_96_NONCE_LENGTH + encryptedBytes.length];
             System.arraycopy(nonce, 0, combinedBytes, 0, GCM_96_NONCE_LENGTH);
@@ -73,7 +81,7 @@ public class AES128GCM96 implements EncryptionAlgorithm {
             System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH);
             System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length);
             GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce);
-            cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec);
+            cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
             byte[] decryptedBytes = cipher.doFinal(encryptedBytes);
             decryptedString = new String(decryptedBytes);
         } catch (Exception e) {
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java
index 0306616..c3dd83c 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/AES256GCM96.java
@@ -1,14 +1,18 @@
 package com.geedgenetworks.core.udf.encrypt;
 
 import cn.hutool.core.util.RandomUtil;
-import com.geedgenetworks.core.pojo.KmsKey;
+import com.geedgenetworks.common.exception.CommonErrorCode;
+import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
 
 import javax.crypto.Cipher;
+import javax.crypto.NoSuchPaddingException;
 import javax.crypto.spec.GCMParameterSpec;
 import javax.crypto.spec.SecretKeySpec;
+import java.security.NoSuchAlgorithmException;
 import java.util.Base64;
 
-public class AES256GCM96 implements EncryptionAlgorithm {
+public class AES256GCM96 implements Crypto {
     private static final String IDENTIFIER = "aes-256-gcm96";
     private static final String ALGORITHM = "AES";
     private static final String TRANSFORMATION = "AES/GCM/NoPadding";
@@ -18,11 +22,15 @@ public class AES256GCM96 implements EncryptionAlgorithm {
     private static final byte[] DEFAULT_SECRET_KEY = ".........geedgenetworks.........".getBytes();
 
     private final Cipher cipher;
-    private KmsKey kmsKey;
+    private DataEncryptionKey dek;
 
-    public AES256GCM96() throws Exception {
-        this.cipher = Cipher.getInstance(TRANSFORMATION);
-        this.kmsKey = new KmsKey(DEFAULT_SECRET_KEY, 1);
+    public AES256GCM96() {
+        try {
+            this.cipher = Cipher.getInstance(TRANSFORMATION);
+            this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1);
+        } catch (NoSuchAlgorithmException | NoSuchPaddingException e) {
+            throw new GrootStreamRuntimeException(CommonErrorCode.UNSUPPORTED_OPERATION, "Failed to initialize AES128GCM96", e);
+        }
     }
 
     @Override
@@ -36,13 +44,13 @@ public class AES256GCM96 implements EncryptionAlgorithm {
     }
 
     @Override
-    public KmsKey getKmsKey() {
-        return kmsKey;
+    public DataEncryptionKey getDataEncryptionKey() {
+        return dek;
     }
 
     @Override
-    public void setKmsKey(KmsKey kmsKey) {
-        this.kmsKey = kmsKey;
+    public void setDataEncryptionKey(DataEncryptionKey dek) {
+        this.dek = dek;
     }
 
     @Override
@@ -51,7 +59,7 @@ public class AES256GCM96 implements EncryptionAlgorithm {
         try {
             byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH);
             GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce);
-            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec);
+            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
             byte[] encryptedBytes = cipher.doFinal(content.getBytes());
             byte[] combinedBytes = new byte[GCM_96_NONCE_LENGTH + encryptedBytes.length];
             System.arraycopy(nonce, 0, combinedBytes, 0, GCM_96_NONCE_LENGTH);
@@ -73,7 +81,7 @@ public class AES256GCM96 implements EncryptionAlgorithm {
             System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH);
             System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length);
             GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce);
-            cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec);
+            cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
             byte[] decryptedBytes = cipher.doFinal(encryptedBytes);
             decryptedString = new String(decryptedBytes);
         } catch (Exception e) {
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/EncryptionAlgorithm.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/Crypto.java
index 3fc4e74..43d7dce 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/EncryptionAlgorithm.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/Crypto.java
@@ -1,15 +1,15 @@
 package com.geedgenetworks.core.udf.encrypt;
 
-import com.geedgenetworks.core.pojo.KmsKey;
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
 
-public interface EncryptionAlgorithm {
+public interface Crypto {
     String getIdentifier();
 
     int getSecretKeyLength();
 
-    KmsKey getKmsKey();
+    DataEncryptionKey getDataEncryptionKey();
 
-    void setKmsKey(KmsKey kmsKey);
+    void setDataEncryptionKey(DataEncryptionKey dek);
 
     String encrypt(String content);
 
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java
index f4ad0a2..d6d701c 100644
--- a/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java
+++ b/groot-core/src/main/java/com/geedgenetworks/core/udf/encrypt/SM4GCM96.java
@@ -1,14 +1,18 @@
 package com.geedgenetworks.core.udf.encrypt;
 
 import cn.hutool.core.util.RandomUtil;
-import com.geedgenetworks.core.pojo.KmsKey;
+import com.geedgenetworks.common.exception.CommonErrorCode;
+import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
 
 import javax.crypto.Cipher;
+import javax.crypto.NoSuchPaddingException;
 import javax.crypto.spec.GCMParameterSpec;
 import javax.crypto.spec.SecretKeySpec;
+import java.security.NoSuchAlgorithmException;
 import java.util.Base64;
 
-public class SM4GCM96 implements EncryptionAlgorithm {
+public class SM4GCM96 implements Crypto {
     private static final String IDENTIFIER = "sm4-gcm96";
     private static final String ALGORITHM = "SM4";
     private static final String TRANSFORMATION = "SM4/GCM/NoPadding";
@@ -18,11 +22,15 @@ public class SM4GCM96 implements EncryptionAlgorithm {
     private static final byte[] DEFAULT_SECRET_KEY = ".geedgenetworks.".getBytes();
 
     private final Cipher cipher;
-    private KmsKey kmsKey;
+    private DataEncryptionKey dek;
 
-    public SM4GCM96() throws Exception {
-        this.cipher = Cipher.getInstance(TRANSFORMATION);
-        this.kmsKey = new KmsKey(DEFAULT_SECRET_KEY, 1);
+    public SM4GCM96() {
+        try {
+            this.cipher = Cipher.getInstance(TRANSFORMATION);
+            this.dek = new DataEncryptionKey(DEFAULT_SECRET_KEY, 1);
+        } catch (NoSuchAlgorithmException | NoSuchPaddingException e) {
+            throw new GrootStreamRuntimeException (CommonErrorCode.UNSUPPORTED_OPERATION, "Failed to initialize SM4GCM96", e);
+        }
     }
 
     @Override
@@ -36,13 +44,13 @@ public class SM4GCM96 implements EncryptionAlgorithm {
     }
 
     @Override
-    public KmsKey getKmsKey() {
-        return kmsKey;
+    public DataEncryptionKey getDataEncryptionKey() {
+        return dek;
     }
 
     @Override
-    public void setKmsKey(KmsKey kmsKey) {
-        this.kmsKey = kmsKey;
+    public void setDataEncryptionKey(DataEncryptionKey dek) {
+        this.dek = dek;
     }
 
     @Override
@@ -51,7 +59,7 @@ public class SM4GCM96 implements EncryptionAlgorithm {
         try {
             byte[] nonce = RandomUtil.randomBytes(GCM_96_NONCE_LENGTH);
             GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce);
-            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec);
+            cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
             byte[] encryptedBytes = cipher.doFinal(content.getBytes());
             byte[] combinedBytes = new byte[GCM_96_NONCE_LENGTH + encryptedBytes.length];
             System.arraycopy(nonce, 0, combinedBytes, 0, GCM_96_NONCE_LENGTH);
@@ -73,7 +81,7 @@ public class SM4GCM96 implements EncryptionAlgorithm {
             System.arraycopy(combined, 0, nonce, 0, GCM_96_NONCE_LENGTH);
             System.arraycopy(combined, GCM_96_NONCE_LENGTH, encryptedBytes, 0, encryptedBytes.length);
             GCMParameterSpec gcmSpec = new GCMParameterSpec(GCM_TAG_LENGTH, nonce);
-            cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(kmsKey.getKeyData(), ALGORITHM), gcmSpec);
+            cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dek.getData(), ALGORITHM), gcmSpec);
             byte[] decryptedBytes = cipher.doFinal(encryptedBytes);
             decryptedString = new String(decryptedBytes);
         } catch (Exception e) {
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
index a8941e2..0f8c851 100644
--- 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
@@ -13,8 +13,8 @@ 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"));
-
+    NAMESPACE_SUBSCRIBER(UUID.fromString("6ba7b893-9dad-11d1-80b4-00c04fd430c8")),
+    NAMESPACE_CELL(UUID.fromString("6ba7b894-9dad-11d1-80b4-00c04fd430c8"));
     private final UUID uuid;
 
     // Static map to hold the mapping from name to UUID
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/CryptoProvider.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/CryptoProvider.java
new file mode 100644
index 0000000..4cc3c25
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/CryptoProvider.java
@@ -0,0 +1,61 @@
+package com.geedgenetworks.core.utils;
+
+import com.geedgenetworks.core.udf.encrypt.Crypto;
+import com.geedgenetworks.core.udf.encrypt.AES128GCM;
+import com.geedgenetworks.core.udf.encrypt.AES128GCM96;
+import com.geedgenetworks.core.udf.encrypt.AES256GCM96;
+import com.geedgenetworks.core.udf.encrypt.SM4GCM96;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+
+@Slf4j
+public final class CryptoProvider {
+
+    @Getter
+    public enum Algorithm {
+        AES_128_GCM("aes-128-gcm"),
+        AES_128_GCM96("aes-128-gcm96"),
+        AES_256_GCM96("aes-256-gcm96"),
+        SM4_GCM96("sm4-gcm96");
+
+        private final String identifier;
+
+        Algorithm(String identifier) {
+            this.identifier = identifier;
+        }
+
+        public static Algorithm fromIdentifier(String identifier) {
+            for (Algorithm algorithm : values()) {
+                if (algorithm.identifier.equalsIgnoreCase(identifier)) {
+                    return algorithm;
+                }
+            }
+            throw new IllegalArgumentException("Unsupported algorithm identifier: " + identifier);
+        }
+    }
+
+    private static final Map<String, Supplier<Crypto>> ALGORITHM_REGISTRY = new HashMap<>();
+
+    static {
+        ALGORITHM_REGISTRY.put(Algorithm.AES_128_GCM.getIdentifier(), AES128GCM::new);
+        ALGORITHM_REGISTRY.put(Algorithm.AES_128_GCM96.getIdentifier(), AES128GCM96::new);
+        ALGORITHM_REGISTRY.put(Algorithm.AES_256_GCM96.getIdentifier(), AES256GCM96::new);
+        ALGORITHM_REGISTRY.put(Algorithm.SM4_GCM96.getIdentifier(), SM4GCM96::new);
+    }
+
+    private CryptoProvider() {
+        // Private constructor to prevent instantiation
+    }
+
+    public static Crypto createEncryptionAlgorithm(String identifier) {
+        Supplier<Crypto> supplier = ALGORITHM_REGISTRY.get(identifier.toLowerCase());
+        if (supplier == null) {
+            throw new IllegalArgumentException("Unsupported algorithm identifier: " + identifier);
+        }
+        return supplier.get();  // Lazily creates a new instance when called
+    }
+}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/EncryptionAlgorithmUtils.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/EncryptionAlgorithmUtils.java
deleted file mode 100644
index 0327e49..0000000
--- a/groot-core/src/main/java/com/geedgenetworks/core/utils/EncryptionAlgorithmUtils.java
+++ /dev/null
@@ -1,30 +0,0 @@
-package com.geedgenetworks.core.utils;
-
-import com.geedgenetworks.core.udf.encrypt.EncryptionAlgorithm;
-import com.geedgenetworks.core.udf.encrypt.AES128GCM96;
-import com.geedgenetworks.core.udf.encrypt.AES256GCM96;
-import com.geedgenetworks.core.udf.encrypt.SM4GCM96;
-import lombok.extern.slf4j.Slf4j;
-
-/**
- * Crypto shade utilities
- */
-@Slf4j
-public final class EncryptionAlgorithmUtils {
-    public static final String ALGORITHM_AES_128_GCM96_NAME = "aes-128-gcm96";
-    public static final String ALGORITHM_AES_256_GCM96_NAME = "aes-256-gcm96";
-    public static final String ALGORITHM_SM4_GCM96_NAME = "sm4-gcm96";
-
-    public static EncryptionAlgorithm createEncryptionAlgorithm(String identifier) throws Exception {
-        switch (identifier) {
-            case ALGORITHM_AES_128_GCM96_NAME:
-                return new AES128GCM96();
-            case ALGORITHM_AES_256_GCM96_NAME:
-                return new AES256GCM96();
-            case ALGORITHM_SM4_GCM96_NAME:
-                return new SM4GCM96();
-            default:
-                return null;
-        }
-    }
-}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/KMSUtils.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/KMSUtils.java
new file mode 100644
index 0000000..f81aae6
--- /dev/null
+++ b/groot-core/src/main/java/com/geedgenetworks/core/utils/KMSUtils.java
@@ -0,0 +1,69 @@
+package com.geedgenetworks.core.utils;
+
+import cn.hutool.core.util.StrUtil;
+import com.geedgenetworks.common.config.KmsConfig;
+import com.geedgenetworks.common.config.ClientSSLConfig;
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
+import io.github.jopenlibs.vault.SslConfig;
+import io.github.jopenlibs.vault.Vault;
+import io.github.jopenlibs.vault.VaultConfig;
+import io.github.jopenlibs.vault.VaultException;
+import io.github.jopenlibs.vault.json.JsonObject;
+import io.github.jopenlibs.vault.response.AuthResponse;
+import io.github.jopenlibs.vault.response.LogicalResponse;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.util.Base64;
+
+@Slf4j
+public class KMSUtils {
+    public static final String KMS_TYPE_LOCAL = "local";
+    public static final String KMS_TYPE_VAULT = "vault";
+
+    public static DataEncryptionKey getVaultKey(KmsConfig kmsConfig, ClientSSLConfig sslConfig, String identifier) {
+        try {
+            Vault vault = getVaultClient(kmsConfig, sslConfig);
+            String exportKeyPath = constructKeyPath(kmsConfig, identifier);
+            LogicalResponse exportResponse = vault.logical().read(exportKeyPath);
+            if (exportResponse.getRestResponse().getStatus() == 200) {
+                JsonObject keys = exportResponse.getDataObject().get("keys").asObject();
+                return new DataEncryptionKey(Base64.getDecoder().decode(StrUtil.trim(keys.get(keys.size() + "").asString(), '"')), keys.size());
+            }
+        } catch (VaultException e) {
+            log.error("Error retrieving vault key from path: {}. ", identifier, e);
+        } catch (Exception e) {
+            log.error("Unexpected error retrieving vault key.", e);
+        }
+        return null;
+    }
+
+    private static String constructKeyPath(KmsConfig kmsConfig, String identifier) {
+        return (CryptoProvider.Algorithm.SM4_GCM96.getIdentifier().equals(identifier)
+                ? kmsConfig.getPluginKeyPath()
+                : kmsConfig.getDefaultKeyPath()) + "/export/encryption-key/" + identifier;
+    }
+
+    public static Vault getVaultClient(KmsConfig kmsConfig, ClientSSLConfig sslConfig) throws VaultException {
+        VaultConfig config = new VaultConfig()
+                .address(kmsConfig.getUrl())
+                .engineVersion(1)
+                .sslConfig(configureSSL(sslConfig))
+                .build();
+        AuthResponse authResponse = Vault.create(config).auth().loginByUserPass(kmsConfig.getUsername(), kmsConfig.getPassword());
+        config.token(authResponse.getAuthClientToken());
+        return Vault.create(config);
+    }
+
+    public static SslConfig configureSSL(ClientSSLConfig clientSSLConfig) throws VaultException {
+        boolean verifySSL = clientSSLConfig != null && !clientSSLConfig.getSkipVerification();
+        SslConfig sslConfig = new SslConfig().verify(verifySSL).build();
+        if (verifySSL) {
+            sslConfig.pemFile(new File(clientSSLConfig.getCaCertificatePath()))
+                    .clientPemFile(new File(clientSSLConfig.getCertificatePath()))
+                    .clientKeyPemFile(new File(clientSSLConfig.getPrivateKeyPath()))
+                    .build();
+        }
+        return sslConfig;
+    }
+}
diff --git a/groot-core/src/main/java/com/geedgenetworks/core/utils/KmsUtils.java b/groot-core/src/main/java/com/geedgenetworks/core/utils/KmsUtils.java
deleted file mode 100644
index 9519dd5..0000000
--- a/groot-core/src/main/java/com/geedgenetworks/core/utils/KmsUtils.java
+++ /dev/null
@@ -1,71 +0,0 @@
-package com.geedgenetworks.core.utils;
-
-import cn.hutool.core.util.StrUtil;
-import com.geedgenetworks.common.config.KmsConfig;
-import com.geedgenetworks.common.config.SSLConfig;
-import com.geedgenetworks.core.pojo.KmsKey;
-import io.github.jopenlibs.vault.SslConfig;
-import io.github.jopenlibs.vault.Vault;
-import io.github.jopenlibs.vault.VaultConfig;
-import io.github.jopenlibs.vault.VaultException;
-import io.github.jopenlibs.vault.json.JsonObject;
-import io.github.jopenlibs.vault.response.AuthResponse;
-import io.github.jopenlibs.vault.response.LogicalResponse;
-import lombok.extern.slf4j.Slf4j;
-
-import java.io.File;
-import java.util.Base64;
-
-@Slf4j
-public class KmsUtils {
-    public static final String KMS_TYPE_LOCAL = "local";
-    public static final String KMS_TYPE_VAULT = "vault";
-
-    public static KmsKey getVaultKey(KmsConfig kmsConfig, SSLConfig sslConfig, String identifier) throws Exception {
-        Vault vault = getVaultClient(kmsConfig, sslConfig);
-        String exportKeyPath;
-        if (EncryptionAlgorithmUtils.ALGORITHM_SM4_GCM96_NAME.equals(identifier)) {
-            exportKeyPath = kmsConfig.getPluginKeyPath() + "/export/encryption-key/" + identifier;
-        } else {
-            exportKeyPath = kmsConfig.getDefaultKeyPath() + "/export/encryption-key/" + identifier;
-        }
-        LogicalResponse exportResponse = vault.logical().read(exportKeyPath);
-        if (exportResponse.getRestResponse().getStatus() == 200) {
-            JsonObject keys = exportResponse.getDataObject().get("keys").asObject();
-            return new KmsKey(Base64.getDecoder().decode(StrUtil.trim(keys.get(keys.size() + "").asString(), '"')), keys.size());
-        } else {
-            throw new RuntimeException("Get vault key error! code: " + exportResponse.getRestResponse().getStatus() + " body: " + new String(exportResponse.getRestResponse().getBody()));
-        }
-    }
-
-    public static Vault getVaultClient(KmsConfig kmsConfig, SSLConfig sslConfig) throws VaultException {
-        String username = kmsConfig.getUsername();
-        String password = kmsConfig.getPassword();
-        String url = kmsConfig.getUrl();
-        boolean skipVerification = true;
-        String caCertificatePath = null;
-        String certificatePath = null;
-        String privateKeyPath = null;
-        if (sslConfig != null) {
-            skipVerification = sslConfig.getSkipVerification();
-            caCertificatePath = sslConfig.getCaCertificatePath();
-            certificatePath = sslConfig.getCertificatePath();
-            privateKeyPath = sslConfig.getPrivateKeyPath();
-        }
-        SslConfig vaultSslConfig = new SslConfig().verify(!skipVerification).build();
-        if (!skipVerification) {
-            vaultSslConfig.pemFile(new File(caCertificatePath))
-                    .clientPemFile(new File(certificatePath))
-                    .clientKeyPemFile(new File(privateKeyPath))
-                    .build();
-        }
-        VaultConfig config = new VaultConfig()
-                .address(url)
-                .engineVersion(1)
-                .sslConfig(vaultSslConfig)
-                .build();
-        AuthResponse authResponse = Vault.create(config).auth().loginByUserPass(username, password);
-        config.token(authResponse.getAuthClientToken());
-        return Vault.create(config);
-    }
-}
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/EncryptFunctionTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/EncryptFunctionTest.java
index e9f1698..20f3c0d 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/EncryptFunctionTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/EncryptFunctionTest.java
@@ -6,15 +6,16 @@ import com.geedgenetworks.common.Constants;
 import com.geedgenetworks.common.Event;
 import com.geedgenetworks.common.config.CommonConfig;
 import com.geedgenetworks.common.config.KmsConfig;
-import com.geedgenetworks.common.config.SSLConfig;
+import com.geedgenetworks.common.config.ClientSSLConfig;
 import com.geedgenetworks.common.exception.GrootStreamRuntimeException;
 import com.geedgenetworks.common.udf.UDFContext;
-import com.geedgenetworks.core.pojo.KmsKey;
+import com.geedgenetworks.core.pojo.DataEncryptionKey;
 import com.geedgenetworks.core.udf.Encrypt;
-import com.geedgenetworks.core.udf.encrypt.EncryptionAlgorithm;
-import com.geedgenetworks.core.utils.EncryptionAlgorithmUtils;
+import com.geedgenetworks.core.udf.encrypt.Crypto;
+import com.geedgenetworks.core.utils.CryptoProvider;
 import com.geedgenetworks.core.utils.HttpClientPoolUtil;
-import com.geedgenetworks.core.utils.KmsUtils;
+import com.geedgenetworks.core.utils.KMSUtils;
+import io.github.jopenlibs.vault.VaultException;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.configuration.Configuration;
@@ -55,11 +56,11 @@ public class EncryptFunctionTest {
     @Test
     public void testEncryptByVault() throws Exception {
         String secretKey = RandomUtil.randomString(32);
-        MockedStatic<KmsUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KmsUtils.class);
-        Mockito.when(KmsUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new KmsKey(secretKey.getBytes(), 1));
+        MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class);
+        Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1));
         RuntimeContext runtimeContext = mockVaultRuntimeContext();
         Map<String, Object> map = new HashMap<>();
-        map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME);
+        map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
         udfContext.setParameters(map);
         Encrypt encrypt = new Encrypt();
         encrypt.open(runtimeContext, udfContext);
@@ -68,12 +69,12 @@ public class EncryptFunctionTest {
         extractedFields.put("phone_number", DATA);
         event.setExtractedFields(extractedFields);
         Event result = encrypt.evaluate(event);
-        EncryptionAlgorithm encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME);
-        assertNotNull(encryptionAlgorithm);
-        encryptionAlgorithm.setKmsKey(new KmsKey(secretKey.getBytes(), 1));
+        Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        assertNotNull(crypto);
+        crypto.setDataEncryptionKey(new DataEncryptionKey(secretKey.getBytes(), 1));
         String encrypted = result.getExtractedFields().get("phone_number").toString();
         assertTrue(encrypted.contains("vault:v1:"));
-        String decrypted = encryptionAlgorithm.decrypt(encrypted.split(":")[2]);
+        String decrypted = crypto.decrypt(encrypted.split(":")[2]);
         assertEquals(DATA, decrypted);
         encrypt.close();
         kmsUtilsMockedStatic.close();
@@ -81,10 +82,9 @@ public class EncryptFunctionTest {
 
     @Test
     public void testEncryptByLocal() throws Exception {
-        byte[] secretKey = ".........geedgenetworks.........".getBytes();
         RuntimeContext runtimeContext = mockLocalRuntimeContext();
         Map<String, Object> map = new HashMap<>();
-        map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME);
+        map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
         udfContext.setParameters(map);
         Encrypt encrypt = new Encrypt();
         encrypt.open(runtimeContext, udfContext);
@@ -93,61 +93,182 @@ public class EncryptFunctionTest {
         extractedFields.put("phone_number", DATA);
         event.setExtractedFields(extractedFields);
         Event result = encrypt.evaluate(event);
-        EncryptionAlgorithm encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME);
+        Crypto encryptionAlgorithm = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
         assertNotNull(encryptionAlgorithm);
-        encryptionAlgorithm.setKmsKey(new KmsKey(secretKey, 1));
         String decrypted = encryptionAlgorithm.decrypt((String) result.getExtractedFields().get("phone_number"));
         assertEquals(DATA, decrypted);
         encrypt.close();
     }
 
     @Test
+    public void testEncryptGetKeyAndSensitiveFieldsError() throws Exception {
+        String secretKey = RandomUtil.randomString(32);
+        MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class);
+        Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1));
+        RuntimeContext runtimeContext = mockVaultRuntimeContext();
+        Map<String, Object> map = new HashMap<>();
+        map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        udfContext.setParameters(map);
+        Encrypt encrypt = new Encrypt();
+        encrypt.open(runtimeContext, udfContext);
+        Event event = new Event();
+        Map<String, Object> extractedFields = new HashMap<>();
+        extractedFields.put("phone_number", DATA);
+        event.setExtractedFields(extractedFields);
+        Event result = encrypt.evaluate(event);
+        Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        assertNotNull(crypto);
+        crypto.setDataEncryptionKey(new DataEncryptionKey(secretKey.getBytes(), 1));
+        String encrypted = result.getExtractedFields().get("phone_number").toString();
+        assertTrue(encrypted.contains("vault:v1:"));
+        String decrypted = crypto.decrypt(encrypted.split(":")[2]);
+        assertEquals(DATA, decrypted);
+
+        Thread.sleep(90000);
+        event = new Event();
+        extractedFields = new HashMap<>();
+        extractedFields.put("phone_number", DATA);
+        event.setExtractedFields(extractedFields);
+        result = encrypt.evaluate(event);
+        encrypted = result.getExtractedFields().get("phone_number").toString();
+        assertTrue(encrypted.contains("vault:v1:"));
+        decrypted = crypto.decrypt(encrypted.split(":")[2]);
+        assertEquals(DATA, decrypted);
+        encrypt.close();
+        kmsUtilsMockedStatic.close();
+    }
+
+    @Test
+    public void testEncryptNoSensitiveFields() throws Exception {
+        String secretKey = RandomUtil.randomString(32);
+        MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class);
+        Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1));
+        RuntimeContext runtimeContext = mockVaultNoSensitiveFieldsRuntimeContext();
+        Map<String, Object> map = new HashMap<>();
+        map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        udfContext.setParameters(map);
+        Encrypt encrypt = new Encrypt();
+        encrypt.open(runtimeContext, udfContext);
+        Event event = new Event();
+        Map<String, Object> extractedFields = new HashMap<>();
+        extractedFields.put("phone_number", DATA);
+        event.setExtractedFields(extractedFields);
+        Event result = encrypt.evaluate(event);
+        Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        assertNotNull(crypto);
+        crypto.setDataEncryptionKey(new DataEncryptionKey(secretKey.getBytes(), 1));
+        String encrypted = result.getExtractedFields().get("phone_number").toString();
+        assertTrue(encrypted.contains("vault:v1:"));
+        String decrypted = crypto.decrypt(encrypted.split(":")[2]);
+        assertEquals(DATA, decrypted);
+        encrypt.close();
+        kmsUtilsMockedStatic.close();
+    }
+
+    @Test
+    public void testEncryptFirstGetKeyError() throws Exception {
+        RuntimeContext runtimeContext = mockVaultRuntimeContext();
+        Map<String, Object> map = new HashMap<>();
+        map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        udfContext.setParameters(map);
+        Encrypt encrypt = new Encrypt();
+        assertThrows(GrootStreamRuntimeException.class, () -> encrypt.open(runtimeContext, udfContext));
+        encrypt.close();
+    }
+
+    @Test
+    public void testEncryptFirstGetSensitiveFieldsError() throws Exception {
+        httpClientPoolUtilMockedStatic.close();
+        String secretKey = RandomUtil.randomString(32);
+        MockedStatic<KMSUtils> kmsUtilsMockedStatic = Mockito.mockStatic(KMSUtils.class);
+        Mockito.when(KMSUtils.getVaultKey(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new DataEncryptionKey(secretKey.getBytes(), 1));
+        RuntimeContext runtimeContext = mockVaultRuntimeContext();
+        Map<String, Object> map = new HashMap<>();
+        map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        udfContext.setParameters(map);
+        Encrypt encrypt = new Encrypt();
+        assertThrows(GrootStreamRuntimeException.class, () -> encrypt.open(runtimeContext, udfContext));
+        encrypt.close();
+        kmsUtilsMockedStatic.close();
+        httpClientPoolUtilMockedStatic = mockSensitiveFields();
+    }
+
+    @Test
+    public void testEncryptByVaultBySSL() throws Exception {
+        ClientSSLConfig sslConfig = new ClientSSLConfig();
+        sslConfig.setSkipVerification(true);
+        assertDoesNotThrow(() -> KMSUtils.configureSSL(sslConfig));
+
+        sslConfig.setSkipVerification(false);
+        assertThrows(VaultException.class, () -> KMSUtils.configureSSL(sslConfig));
+
+        sslConfig.setSkipVerification(false);
+        sslConfig.setCaCertificatePath("src/test/resources/ssl/ca.crt");
+        sslConfig.setCertificatePath("src/test/resources/ssl/server.crt");
+        sslConfig.setPrivateKeyPath("src/test/resources/ssl/server.key");
+        assertDoesNotThrow(() -> KMSUtils.configureSSL(sslConfig));
+    }
+
+    @Test
     public void testEncryptByIdentifier() {
         Map<String, Object> map = new HashMap<>();
-        map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME);
+        map.put("identifier", CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
         udfContext.setParameters(map);
         Encrypt encrypt1 = new Encrypt();
         assertDoesNotThrow(() -> encrypt1.open(mockLocalRuntimeContext(), udfContext));
         encrypt1.close();
 
         Encrypt encrypt2 = new Encrypt();
-        map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_AES_128_GCM96_NAME);
+        map.put("identifier", CryptoProvider.Algorithm.AES_128_GCM96.getIdentifier());
         udfContext.setParameters(map);
         assertDoesNotThrow(() -> encrypt2.open(mockLocalRuntimeContext(), udfContext));
         encrypt2.close();
 
         Encrypt encrypt3 = new Encrypt();
-        map.put("identifier", EncryptionAlgorithmUtils.ALGORITHM_SM4_GCM96_NAME);
+        map.put("identifier", CryptoProvider.Algorithm.SM4_GCM96.getIdentifier());
         udfContext.setParameters(map);
         assertDoesNotThrow(() -> encrypt3.open(mockLocalRuntimeContext(), udfContext));
         encrypt3.close();
+
+        Encrypt encrypt4 = new Encrypt();
+        map.put("identifier", CryptoProvider.Algorithm.AES_128_GCM.getIdentifier());
+        udfContext.setParameters(map);
+        assertDoesNotThrow(() -> encrypt4.open(mockLocalRuntimeContext(), udfContext));
+        encrypt4.close();
     }
 
     @Test
     public void testEncryptionAlgorithm() throws Exception {
-        EncryptionAlgorithm encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_128_GCM96_NAME);
-        assertNotNull(encryptionAlgorithm);
-        encryptionAlgorithm.setKmsKey(new KmsKey("aaaaaaaaaaaaaaaa".getBytes(), 1));
-        String encryptData = encryptionAlgorithm.encrypt(DATA);
-        String decryptData = encryptionAlgorithm.decrypt(encryptData);
+        Crypto crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_128_GCM96.getIdentifier());
+        assertNotNull(crypto);
+        crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaa".getBytes(), 1));
+        String encryptData = crypto.encrypt(DATA);
+        String decryptData = crypto.decrypt(encryptData);
         assertEquals(DATA, decryptData);
 
-        encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_AES_256_GCM96_NAME);
-        assertNotNull(encryptionAlgorithm);
-        encryptionAlgorithm.setKmsKey(new KmsKey("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".getBytes(), 1));
-        encryptData = encryptionAlgorithm.encrypt(DATA);
-        decryptData = encryptionAlgorithm.decrypt(encryptData);
+        crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_256_GCM96.getIdentifier());
+        assertNotNull(crypto);
+        crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".getBytes(), 1));
+        encryptData = crypto.encrypt(DATA);
+        decryptData = crypto.decrypt(encryptData);
         assertEquals(DATA, decryptData);
 
-        encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm(EncryptionAlgorithmUtils.ALGORITHM_SM4_GCM96_NAME);
-        assertNotNull(encryptionAlgorithm);
-        encryptionAlgorithm.setKmsKey(new KmsKey("aaaaaaaaaaaaaaaa".getBytes(), 1));
-        encryptData = encryptionAlgorithm.encrypt(DATA);
-        decryptData = encryptionAlgorithm.decrypt(encryptData);
+        crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.SM4_GCM96.getIdentifier());
+        assertNotNull(crypto);
+        crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaa".getBytes(), 1));
+        encryptData = crypto.encrypt(DATA);
+        decryptData = crypto.decrypt(encryptData);
+        assertEquals(DATA, decryptData);
+
+        crypto = CryptoProvider.createEncryptionAlgorithm(CryptoProvider.Algorithm.AES_128_GCM.getIdentifier());
+        assertNotNull(crypto);
+        crypto.setDataEncryptionKey(new DataEncryptionKey("aaaaaaaaaaaaaaaa".getBytes(), 1));
+        encryptData = crypto.encrypt(DATA);
+        assertEquals("R2FsYXh5MjAxOSMq6Q4PFGRvBmtSQ36Ug9XDHyMXB7Oye/OPITNW", encryptData);
+        decryptData = crypto.decrypt(encryptData);
         assertEquals(DATA, decryptData);
 
-        encryptionAlgorithm = EncryptionAlgorithmUtils.createEncryptionAlgorithm("sm4");
-        assertNull(encryptionAlgorithm);
+        assertThrows(IllegalArgumentException.class, () -> CryptoProvider.createEncryptionAlgorithm("sm4"));
     }
 
     @Test
@@ -178,12 +299,12 @@ public class EncryptFunctionTest {
         CommonConfig commonConfig = new CommonConfig();
         Map<String, KmsConfig> kmsConfigs = new HashMap<>();
         KmsConfig kmsConfig = new KmsConfig();
-        kmsConfig.setType(KmsUtils.KMS_TYPE_LOCAL);
-        kmsConfigs.put(KmsUtils.KMS_TYPE_LOCAL, kmsConfig);
+        kmsConfig.setType(KMSUtils.KMS_TYPE_LOCAL);
+        kmsConfigs.put(KMSUtils.KMS_TYPE_LOCAL, kmsConfig);
         kmsConfig = new KmsConfig();
-        kmsConfig.setType(KmsUtils.KMS_TYPE_VAULT);
-        kmsConfigs.put(KmsUtils.KMS_TYPE_VAULT, kmsConfig);
-        SSLConfig sslConfig = new SSLConfig();
+        kmsConfig.setType(KMSUtils.KMS_TYPE_VAULT);
+        kmsConfigs.put(KMSUtils.KMS_TYPE_VAULT, kmsConfig);
+        ClientSSLConfig sslConfig = new ClientSSLConfig();
         sslConfig.setSkipVerification(true);
         Map<String, String> propertiesConfig = new HashMap<>();
         propertiesConfig.put("projection.encrypt.schema.registry.uri", "127.0.0.1:9999/v1/schema/session_record?option=encrypt_fields");
@@ -191,7 +312,7 @@ public class EncryptFunctionTest {
         commonConfig.setSslConfig(sslConfig);
         commonConfig.setPropertiesConfig(propertiesConfig);
         configuration.setString(Constants.SYSPROP_GROOTSTREAM_CONFIG, JSON.toJSONString(commonConfig));
-        configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KmsUtils.KMS_TYPE_LOCAL);
+        configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KMSUtils.KMS_TYPE_LOCAL);
         Mockito.when(executionConfig.getGlobalJobParameters()).thenReturn(configuration);
         return runtimeContext;
     }
@@ -208,20 +329,53 @@ public class EncryptFunctionTest {
         CommonConfig commonConfig = new CommonConfig();
         Map<String, KmsConfig> kmsConfigs = new HashMap<>();
         KmsConfig kmsConfig = new KmsConfig();
-        kmsConfig.setType(KmsUtils.KMS_TYPE_VAULT);
-        kmsConfigs.put(KmsUtils.KMS_TYPE_VAULT, kmsConfig);
+        kmsConfig.setType(KMSUtils.KMS_TYPE_VAULT);
+        kmsConfigs.put(KMSUtils.KMS_TYPE_VAULT, kmsConfig);
         kmsConfig = new KmsConfig();
-        kmsConfig.setType(KmsUtils.KMS_TYPE_LOCAL);
-        kmsConfigs.put(KmsUtils.KMS_TYPE_LOCAL, kmsConfig);
-        SSLConfig sslConfig = new SSLConfig();
+        kmsConfig.setType(KMSUtils.KMS_TYPE_LOCAL);
+        kmsConfigs.put(KMSUtils.KMS_TYPE_LOCAL, kmsConfig);
+        ClientSSLConfig sslConfig = new ClientSSLConfig();
         sslConfig.setSkipVerification(true);
         Map<String, String> propertiesConfig = new HashMap<>();
         propertiesConfig.put("projection.encrypt.schema.registry.uri", "127.0.0.1:9999/v1/schema/session_record?option=encrypt_fields");
+        propertiesConfig.put(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "1");
+        propertiesConfig.put(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "1");
+        commonConfig.setKmsConfig(kmsConfigs);
+        commonConfig.setSslConfig(sslConfig);
+        commonConfig.setPropertiesConfig(propertiesConfig);
+        configuration.setString(Constants.SYSPROP_GROOTSTREAM_CONFIG, JSON.toJSONString(commonConfig));
+        configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KMSUtils.KMS_TYPE_VAULT);
+        Mockito.when(executionConfig.getGlobalJobParameters()).thenReturn(configuration);
+        return runtimeContext;
+    }
+
+    static RuntimeContext mockVaultNoSensitiveFieldsRuntimeContext() {
+        RuntimeContext runtimeContext = Mockito.mock(RuntimeContext.class);
+        ExecutionConfig executionConfig = Mockito.mock(ExecutionConfig.class);
+        Mockito.when(runtimeContext.getExecutionConfig()).thenReturn(executionConfig);
+        MetricGroup metricGroup = Mockito.mock(OperatorMetricGroup.class);
+        Mockito.when(runtimeContext.getMetricGroup()).thenReturn(metricGroup);
+        Mockito.when(metricGroup.addGroup(Mockito.anyString())).thenReturn(metricGroup);
+        Mockito.when(metricGroup.counter(Mockito.anyString())).thenReturn(new SimpleCounter());
+        Configuration configuration = new Configuration();
+        CommonConfig commonConfig = new CommonConfig();
+        Map<String, KmsConfig> kmsConfigs = new HashMap<>();
+        KmsConfig kmsConfig = new KmsConfig();
+        kmsConfig.setType(KMSUtils.KMS_TYPE_VAULT);
+        kmsConfigs.put(KMSUtils.KMS_TYPE_VAULT, kmsConfig);
+        kmsConfig = new KmsConfig();
+        kmsConfig.setType(KMSUtils.KMS_TYPE_LOCAL);
+        kmsConfigs.put(KMSUtils.KMS_TYPE_LOCAL, kmsConfig);
+        ClientSSLConfig sslConfig = new ClientSSLConfig();
+        sslConfig.setSkipVerification(true);
+        Map<String, String> propertiesConfig = new HashMap<>();
+        propertiesConfig.put(Constants.SYSPROP_ENCRYPT_KMS_KEY_SCHEDULER_INTERVAL_NAME, "1");
+        propertiesConfig.put(Constants.SYSPROP_ENCRYPT_SENSITIVE_FIELDS_SCHEDULER_INTERVAL_NAME, "1");
         commonConfig.setKmsConfig(kmsConfigs);
         commonConfig.setSslConfig(sslConfig);
         commonConfig.setPropertiesConfig(propertiesConfig);
         configuration.setString(Constants.SYSPROP_GROOTSTREAM_CONFIG, JSON.toJSONString(commonConfig));
-        configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KmsUtils.KMS_TYPE_VAULT);
+        configuration.setString(Constants.SYSPROP_KMS_TYPE_CONFIG, KMSUtils.KMS_TYPE_VAULT);
         Mockito.when(executionConfig.getGlobalJobParameters()).thenReturn(configuration);
         return runtimeContext;
     }
diff --git a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java
index 534569b..70216b4 100644
--- a/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java
+++ b/groot-core/src/test/java/com/geedgenetworks/core/udf/test/simple/UUIDTest.java
@@ -136,4 +136,21 @@ public class UUIDTest {
         Event result1 = uuidv5.evaluate(event);
         assertEquals("9b154520-3c29-541c-bb81-f649354dae67", result1.getExtractedFields().get("uuid").toString());
     }
+    @Test
+    public void testUuidV5ForCell() {
+        udfContext = new UDFContext();
+        UUIDv5 uuidv5 = new UUIDv5();
+        parameters =  new HashMap<>();
+        udfContext.setParameters(parameters);
+        udfContext.setLookupFields(List.of("cell"));
+        udfContext.setOutputFields(Collections.singletonList("uuid"));
+        parameters.put("namespace","NAMESPACE_CELL");
+        uuidv5.open(null, udfContext);
+        Event event = new Event();
+        Map<String, Object> extractedFields = new HashMap<>();
+        extractedFields.put("cell", "test1");
+        event.setExtractedFields(extractedFields);
+        Event result1 = uuidv5.evaluate(event);
+        assertEquals("4693c13f-e97f-5846-bdb0-f80ce2d2869d", result1.getExtractedFields().get("uuid").toString());
+    }
 }
diff --git a/groot-core/src/test/resources/ssl/ca.crt b/groot-core/src/test/resources/ssl/ca.crt
new file mode 100644
index 0000000..91201f9
--- /dev/null
+++ b/groot-core/src/test/resources/ssl/ca.crt
@@ -0,0 +1,18 @@
+-----BEGIN CERTIFICATE-----
+MIIC8TCCAdmgAwIBAgIJAOwjrQQ/LLx9MA0GCSqGSIb3DQEBCwUAMA8xDTALBgNV
+BAMMBE15Q0EwHhcNMjQxMDEyMDMxNzU3WhcNMzQxMDEwMDMxNzU3WjAPMQ0wCwYD
+VQQDDARNeUNBMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAqD/Yapxq
+Nm9+4+JccB5XPJ5dTeu7qQORFHBVpPClnLYXsNdmYq+N7ixhsXBKbHdzlB15dZI4
+lTvvGnf7sW4ij1wz42HkF5ZnalP6b7FjxfDqAavFEDtdfOT8f5hkbGrrC2F5GogQ
+BoqwFxqHx19UxO4OU4NHB8pSGpFhFgZ0xDkd0gRE6WKXOoWfDLzalFlKI8ISePjq
+JD4RbsjsM2PF7sQNWl+Vf8JX+BwV8u7IFuN/qdakDIPn9KQ2KBhHYCdOqK8SSoKz
+Pf5TvoaiNGjCLeDG2NuRQV4WmA1Gcs0az4rhnYRk2azlWBJld/yigRE5+xsGl7fK
+gqVA7v6kR+GbswIDAQABo1AwTjAdBgNVHQ4EFgQU1K3YFle/hPPdpMjVLnHitVgr
+v50wHwYDVR0jBBgwFoAU1K3YFle/hPPdpMjVLnHitVgrv50wDAYDVR0TBAUwAwEB
+/zANBgkqhkiG9w0BAQsFAAOCAQEAf7MkFkypRF+e7qt1DUkwS3zPJUYXJOriHp3F
+W+gQs0kDTEF3TRVJAITU4nzveWO77KVsPd8zs1W5sDxGnsIkhlJ3NVsYKone28Rz
+AWaVy48J9Pj4O1hk9GLH0F6vkByRbXC6Cmcu3C7Tvxnxmegni98/Ja/ASAlBMLrE
+YMl5kG87vgAMgYB7RETA9KmzNkTGIz4UcvqN/7RGxnj7bJP0fe/kwlZyliT4wHbQ
+s7+9Jt1kE+fFgpz4q3gq2DEcRenFS43jb53WFPHKD8E1fXhk32pqr9aU+v7wHhKo
+lX9K1dXWvkw+NnMZIFV9VurfuRlngnfd7mD8yEoMK50VR01JSQ==
+-----END CERTIFICATE-----
diff --git a/groot-core/src/test/resources/ssl/server.crt b/groot-core/src/test/resources/ssl/server.crt
new file mode 100644
index 0000000..19d40c7
--- /dev/null
+++ b/groot-core/src/test/resources/ssl/server.crt
@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDUDCCAjigAwIBAgIJAPO4WswvRR16MA0GCSqGSIb3DQEBCwUAMA8xDTALBgNV
+BAMMBE15Q0EwHhcNMjQxMDEyMDMxNzU3WhcNMzQxMDEwMDMxNzU3WjBhMQswCQYD
+VQQGEwJVUzEOMAwGA1UECAwFU3RhdGUxDTALBgNVBAcMBENpdHkxFTATBgNVBAoM
+DE9yZ2FuaXphdGlvbjENMAsGA1UECwwEVW5pdDENMAsGA1UEAwwETXlDQTCCASIw
+DQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKrUoeeyulohpHkmawWMJ6j44hES
+SzxS83cmqPpwBMksW60HQ+lQXalDYERdwGbgrN/v2YcYQx3NqfnE9UDXal7QQYp6
+d+TpVw+LqQOzinzxsC/kJp4RRcIkiGYyE6EedYFVzVOeQiB5HogG3LPU/CdCjlXq
+WhoqAF0jPoGmkUQZiGNfXJx7eWK40C4Bw2q+Q8sKxlLuqHWOMF/dFWWCmFyDMdZ9
+LdQjgYy8s6Bq73FhU5n4fQ+ivGt8iNvJAqJP7wgPGt6UEi+i7hIoWIqeZPFnqj5w
+t3FtLfpK7NBua7YLfIE0J8oA42QplmEh1Mn+l6/tj0K0cYUaDHSbm6ccITcCAwEA
+AaNdMFswCQYDVR0TBAIwADALBgNVHQ8EBAMCBLAwHQYDVR0lBBYwFAYIKwYBBQUH
+AwEGCCsGAQUFBwMCMCIGA1UdEQQbMBmCEXNlcnZlci5kb21haW4uY29thwTAqCjf
+MA0GCSqGSIb3DQEBCwUAA4IBAQBYo2N6Ta93LM8RatQz1fPcD8qnwod+JTINGSYs
+nC4qqMtc3XtMN7ZrZr0IQAS2s7LiDsoIzNDIG3sJTnCKAYrE0rQcTtMHK2uLtntd
+8L4Vjj9upJG0faOtpqgBbTD9BwVXIhhjgttoIMh90pyr62KsK/KYDumWd+yikkkm
+PazsSfm2AURfiANLKk+eq/N4WWPLMJN6HbEzJtnuxYtIRj7VHNjnKD5p8+ulYCzQ
+pehmudmNu6UfzwsuAR/HI8tulnw6v8GS3lz/9yCf8W+DLG0dWE1i5/A5Hjcu78mW
+NDzarO9lXwJhgeXqn4fxq81b7nGIneTvkpR8vhOG1P+HZiwd
+-----END CERTIFICATE-----
diff --git a/groot-core/src/test/resources/ssl/server.key b/groot-core/src/test/resources/ssl/server.key
new file mode 100644
index 0000000..a2efffe
--- /dev/null
+++ b/groot-core/src/test/resources/ssl/server.key
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQCq1KHnsrpaIaR5
+JmsFjCeo+OIREks8UvN3Jqj6cATJLFutB0PpUF2pQ2BEXcBm4Kzf79mHGEMdzan5
+xPVA12pe0EGKenfk6VcPi6kDs4p88bAv5CaeEUXCJIhmMhOhHnWBVc1TnkIgeR6I
+Btyz1PwnQo5V6loaKgBdIz6BppFEGYhjX1yce3liuNAuAcNqvkPLCsZS7qh1jjBf
+3RVlgphcgzHWfS3UI4GMvLOgau9xYVOZ+H0PorxrfIjbyQKiT+8IDxrelBIvou4S
+KFiKnmTxZ6o+cLdxbS36SuzQbmu2C3yBNCfKAONkKZZhIdTJ/pev7Y9CtHGFGgx0
+m5unHCE3AgMBAAECggEAJndMoaSAC62JoHIDJTOi9oxcMyXgZQv0oH7HC+VPXpEr
+b3G0XAPpoyR1t884dLHgm2ghnibbbgmSXZh49QnMfN798xWSi6vzG6ACBcBWzb2K
+Q65m967B+25IfGKIQv5dzSqp2ktHbpJ3Sn/pEGFECf8Vl8j2Uu/kNxSpjX4ZNbD/
+5Nz+gtrqvlOBhr7usuIS8SYEuc3/wpYDmL9cR0ws+Uuc/kjecH/UzuGtuFeai70h
+9/whiM+q5poBCH6Wjp0rYUlPa68O/yROtrJsB7a7BXR1nvFxtn5G6T2eKjQac9xs
+JI+ruUw0RbFGyzRbtKUdBhWRUeYYjBYoq+CcB3odGQKBgQDYwzmQV64Xyfwx/fIw
+wepLlAKNAhqY8z6+1Pg7qhutaeFYG5uv3jWYWpmgWO+WyKefjyPdIzNyU/36OO7/
+s5r3ZryA/15DRlvEDbSDAeVyjQEDNdDIzfNqQ0h+8j7vBJlykGgCekMITjiJUv8a
+2yUsHyBevI3WFpUDaAMUtUmK3QKBgQDJwOtdR+H62/1PX23vyP6c/zU520j92zQv
+5yrHVQINtFOSmzBgdpM/G0yLEJIbw0Fvz8/cx8IYUQnQQPuCS7ZxJJSqiwxV33Qz
+hhtZJhH61lmE2guHURdytaE9heGMwfrgnmG64kdlVNFHZ492ltSIq/C+zUB2/5YT
+Mm0f8yfpIwKBgQDObTzIpXd52DWANmMK4+EIkK/NMY+60QuUGKU9zMYG46piigg9
+99P6f22GMqwYYIahgWOaGQfJfQuF2+pfQN/3c7NY9dkDIGIL1zFtAcVMzdOFBx8J
+3HhPXjwQCQq9/RdU7wjeMyjbJALbZFrlbIV9+zaMgexhUagfUlJ8yhh7UQKBgQCt
+npVtSsTPqq0MtyTWavOhi4X0ah8gRplcd+S6cQ85V+triJ1TBfelIQr3yaTSu27+
+l6lbZ5RCdMqrKqDF+f3g1AgT02EkLQ3EoS27xCVI5VlYGIQ/SKuTDXbaiPIWvX/1
++JZFyyCBtUH73sT42se/bafZqqxFO6Gcl5KNIiVAXQKBgQDYigejeQ81cL72lf7d
+P/n5HU92cDBAt+sF8CJoyaHqcInn0sm2t9Eix/rcMY6da7cyeA8HSdiwPjmFRkn9
+jJrrJZGkjm9zE2+QX7/nFCV5MXCf4216gjkuAIdcWGQFiNNrw8eokxQKn0Y7N5Jh
+AaOyISDAAaqdFO6Nwsixscvd0A==
+-----END PRIVATE KEY-----