Repository: flink-extended/dl-on-flink (language: python)

Issue:
simplify the start processing and support python2 and python3
1.create the user.properties to  simplify startup process, users only need to fill in the configuration file at the beginning
2.support pyhon2 and python3


Discussion / hints:
[![CLA assistant check](https://cla-assistant.io/pull/badge/signed)](https://cla-assistant.io/alibaba/flink-ai-extended?pullRequest=8) <br/>All committers have signed the CLA.

Fix patch:
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/Event.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/Event.java
deleted file mode 100644
index 408da0778..000000000
--- a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/Event.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *   http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package com.aiflow.notification.client;
-
-public class Event {
-	private String key;
-	private String value;
-	private String event_type;
-	private int version;
-	private long create_time;
-	private long id;
-
-	public Event(String key, String value, String event_type) {
-		this.key = key;
-		this.value = value;
-		this.event_type = event_type;
-	}
-
-	public Event(String key, String value, String event_type, int version, long create_time, long id) {
-		this.key = key;
-		this.value = value;
-		this.event_type = event_type;
-		this.version = version;
-		this.create_time = create_time;
-		this.id = id;
-	}
-
-	public String getKey() {
-		return key;
-	}
-
-	public void setKey(String key) {
-		this.key = key;
-	}
-
-	public String getValue() {
-		return value;
-	}
-
-	public void setValue(String value) {
-		this.value = value;
-	}
-
-	public String getEvent_type() {
-		return event_type;
-	}
-
-	public void setEvent_type(String event_type) {
-		this.event_type = event_type;
-	}
-
-	public int getVersion() {
-		return version;
-	}
-
-	public void setVersion(int version) {
-		this.version = version;
-	}
-
-	public long getCreate_time() {
-		return create_time;
-	}
-
-	public void setCreate_time(long create_time) {
-		this.create_time = create_time;
-	}
-
-	public long getId() {
-		return id;
-	}
-
-	public void setId(long id) {
-		this.id = id;
-	}
-}
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java
index 7080a9ee4..19608d691 100644
--- a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java
@@ -18,9 +18,11 @@
  */
 package com.aiflow.notification.client;
 
-import com.aiflow.notification.proto.NotificationServiceGrpc;
+import com.aiflow.notification.entity.EventMeta;
+import com.aiflow.notification.proto.NotificationServiceGrpc.NotificationServiceBlockingStub;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.commons.collections4.CollectionUtils;
+import sun.util.logging.resources.logging;
 
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -29,23 +31,41 @@
 import static com.aiflow.notification.client.NotificationClient.listEvents;
 
 public class EventListener {
-	private NotificationServiceGrpc.NotificationServiceBlockingStub serviceStub;
-	private String key;
-	private Integer version;
-	private EventWatcher watcher;
+
+	private final NotificationServiceBlockingStub serviceStub;
+	private final List<String> keys;
+	private final long version;
+	private final String eventType;
+	private final long startTime;
+	private final String namespace;
+	private final EventWatcher watcher;
 	private final ExecutorService executorService;
-	private int timeoutSeconds;
+	private final int timeoutSeconds;
 	private volatile boolean isRunning = true;
 
-	public EventListener(NotificationServiceGrpc.NotificationServiceBlockingStub serviceStub, String key,
-		Integer version, EventWatcher watcher, Integer timeoutSeconds) {
+	public EventListener(
+			NotificationServiceBlockingStub serviceStub,
+			List<String> keys,
+			long version,
+			String eventType,
+			long startTime,
+			String namespace,
+			EventWatcher watcher,
+			Integer timeoutSeconds) {
 		this.serviceStub = serviceStub;
-		this.key = key;
+		this.keys = keys;
 		this.version = version;
+		this.eventType = eventType;
+		this.startTime = startTime;
+		this.namespace = namespace;
 		this.watcher = watcher;
 		this.timeoutSeconds = timeoutSeconds;
-		this.executorService = Executors.newSingleThreadExecutor(
-			new ThreadFactoryBuilder().setDaemon(true).setNameFormat("listen-notification-%d").build());
+		this.executorService =
+				Executors.newSingleThreadExecutor(
+						new ThreadFactoryBuilder()
+								.setDaemon(true)
+								.setNameFormat("listen-notification-%d")
+								.build());
 	}
 
 	public void start() {
@@ -59,13 +79,22 @@ public void shutdown() {
 
 	public Runnable listenEvents() {
 		return () -> {
-			Integer listenVersion = this.version;
+			long listenVersion = this.version;
 			while (this.isRunning) {
 				try {
 					if (Thread.currentThread().isInterrupted()) {
 						break;
 					}
-					List<Event> events = listEvents(this.serviceStub, this.key, listenVersion, this.timeoutSeconds);
+					List<EventMeta> events =
+							listEvents(
+									this.serviceStub,
+									this.keys,
+									listenVersion,
+									this.eventType,
+									this.startTime,
+									this.namespace,
+									this.timeoutSeconds);
+					System.out.println("######" + events.size());
 					if (CollectionUtils.isNotEmpty(events)) {
 						this.watcher.process(events);
 						listenVersion = events.get(events.size() - 1).getVersion();
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventWatcher.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventWatcher.java
index 7865ca6de..84a64e588 100644
--- a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventWatcher.java
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventWatcher.java
@@ -18,8 +18,11 @@
  */
 package com.aiflow.notification.client;
 
+import com.aiflow.notification.entity.EventMeta;
+
 import java.util.List;
 
 public interface EventWatcher {
-	void process(List<Event> events);
+	void process(List<EventMeta> events);
 }
+
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java
index 7d366367e..86a6c1c6d 100644
--- a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java
@@ -18,145 +18,198 @@
  */
 package com.aiflow.notification.client;
 
-import com.aiflow.notification.proto.NotificationServiceGrpc;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.*;
+import com.aiflow.notification.entity.EventMeta;
+import com.aiflow.notification.proto.NotificationServiceGrpc.NotificationServiceBlockingStub;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.GetLatestVersionByKeyRequest;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.GetLatestVersionResponse;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.ListAllEventsRequest;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.ListEventsRequest;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.ListEventsResponse;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.SendEventRequest;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.SendEventsResponse;
 import io.grpc.ManagedChannelBuilder;
-import com.google.protobuf.util.JsonFormat.Parser;
-import io.grpc.Channel;
 import org.apache.commons.lang3.StringUtils;
+
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
-import static com.google.protobuf.util.JsonFormat.parser;
-
+import static com.aiflow.notification.entity.EventMeta.buildEventMeta;
+import static com.aiflow.notification.proto.NotificationServiceGrpc.newBlockingStub;
 
 public class NotificationClient {
-	public static final String SERVER_URI = "localhost:50051";
-
-	private final NotificationServiceGrpc.NotificationServiceBlockingStub notificationServiceStub;
-	private final Map<String, EventListener> threads;
 
-	private final Parser parser = parser().ignoringUnknownFields();
-
-	public NotificationClient() {
-		this(SERVER_URI);
-	}
+	public static final String SERVER_URI = "localhost:50051";
 
-	public NotificationClient(String target) {
-		this(ManagedChannelBuilder.forTarget(target).usePlaintext().build());
-	}
+	private final String defaultNamespace;
+	private final Boolean enableHa;
+	private final Integer listMemberIntervalMs;
+	private final Integer retryIntervalMs;
+	private final Integer retryTimeoutMs;
+	private final NotificationServiceBlockingStub notificationServiceStub;
+	private final Map<Map<String, String>, EventListener> threads;
 
-	public NotificationClient(Channel channel) {
-		this.notificationServiceStub = NotificationServiceGrpc.newBlockingStub(channel);
+	public NotificationClient(
+			String target,
+			String defaultNamespace,
+			Boolean enableHa,
+			Integer listMemberIntervalMs,
+			Integer retryIntervalMs,
+			Integer retryTimeoutMs) {
+		this.defaultNamespace = defaultNamespace;
+		this.enableHa = enableHa;
+		this.listMemberIntervalMs = listMemberIntervalMs;
+		this.retryIntervalMs = retryIntervalMs;
+		this.retryTimeoutMs = retryTimeoutMs;
+		this.notificationServiceStub =
+				newBlockingStub(
+						ManagedChannelBuilder.forTarget(
+								StringUtils.isEmpty(target) ? SERVER_URI : target)
+								.usePlaintext()
+								.build());
 		this.threads = new HashMap<>();
 	}
 
-	public static Event eventProtoToEvent(EventProto eventProto) {
-		return new Event(eventProto.getKey(), eventProto.getValue(), eventProto.getEventType(),
-			eventProto.getVersion(), eventProto.getCreateTime(), eventProto.getId());
-	}
-
 	/**
 	 * Send the event to Notification Service.
 	 *
 	 * @param key Key of event updated in Notification Service.
 	 * @param value Value of event updated in Notification Service.
+	 * @param eventType Type of event updated in Notification Service.
+	 * @param context Context of event updated in Notification Service.
 	 * @return Object of Event created in Notification Service.
 	 */
-	public Event sendEvent(String key, String value, String eventType) throws Exception {
-		SendEventRequest request = SendEventRequest.newBuilder()
-			.setEvent(EventProto.newBuilder().setKey(key).setValue(value).setEventType(eventType).build()).build();
+	public EventMeta sendEvent(String key, String value, String eventType, String context)
+			throws Exception {
+		SendEventRequest request =
+				SendEventRequest.newBuilder()
+						.setEvent(
+								EventProto.newBuilder()
+										.setKey(key)
+										.setValue(value)
+										.setEventType(eventType)
+										.setContext(context)
+										.setNamespace(defaultNamespace)
+										.build())
+						.setUuid(UUID.randomUUID().toString())
+						.build();
 		SendEventsResponse response = this.notificationServiceStub.sendEvent(request);
-		if (response.getReturnCode().equals(ReturnStatus.ERROR.toString())) {
-			throw new Exception(response.getReturnMsg());
+		if (response.getReturnCode() == ReturnStatus.SUCCESS) {
+			return buildEventMeta(response.getEvent());
 		} else {
-			EventProto eventProto = response.getEvent();
-			return eventProtoToEvent(eventProto);
+			throw new Exception(response.getReturnMsg());
 		}
 	}
 
 	/**
 	 * List specific `key` or `version` notifications in Notification Service.
 	 *
-	 * @param key Key of notification for listening.
+	 * @param keys Keys of notification for listening.
 	 * @param version (Optional) Version of notification for listening.
+	 * @param eventType (Optional) Type of event for listening.
+	 * @param startTime (Optional) Type of event for listening.
 	 * @return List of Notification updated in Notification Service.
 	 */
-	public List<Event> listEvents(String key, int version) throws Exception {
-		return listEvents(this.notificationServiceStub, key, version, 0);
+	public List<EventMeta> listEvents(
+			List<String> keys, long version, String eventType, long startTime) throws Exception {
+		return listEvents(
+				this.notificationServiceStub,
+				keys,
+				version,
+				eventType,
+				startTime,
+				defaultNamespace,
+				0);
 	}
 
 	/**
 	 * List specific registered listener events in Notification Service.
 	 *
 	 * @param serviceStub Notification service GRPC stub.
-	 * @param key Key of event for listening.
+	 * @param keys Keys of event for listening.
 	 * @param version (Optional) Version of event for listening.
 	 * @param timeoutSeconds List events request timeout seconds.
 	 * @return List of event updated in Notification Service.
 	 */
-	public static List<Event> listEvents(
-		NotificationServiceGrpc.NotificationServiceBlockingStub serviceStub, String key, int version,
-		Integer timeoutSeconds) throws Exception {
-		ListEventsRequest request = ListEventsRequest.newBuilder()
-			.setEvent(EventProto.newBuilder().setKey(key)
-				.setVersion(version).build()).setTimeoutSeconds(timeoutSeconds).build();
-		ListEventsResponse response = serviceStub.listEvents(request);
-		return parseEventsFromResponse(response);
+	protected static List<EventMeta> listEvents(
+			NotificationServiceBlockingStub serviceStub,
+			List<String> keys,
+			long version,
+			String eventType,
+			long startTime,
+			String namespace,
+			Integer timeoutSeconds)
+			throws Exception {
+		ListEventsRequest request =
+				ListEventsRequest.newBuilder()
+						.addAllKeys(keys)
+						.setStartVersion(version)
+						.setEventType(eventType)
+						.setStartTime(startTime)
+						.setNamespace(namespace)
+						.setTimeoutSeconds(timeoutSeconds)
+						.build();
+		return parseEventsFromResponse(serviceStub.listEvents(request));
 	}
 
 	/**
+	 * List all registered listener events in Notification Service.
 	 *
-	 * @param startTime The event create time after the given startTime.
+	 * @param startTime (Optional) The event create time after the given startTime.
+	 * @param startVersion (Optional) Start version of event for listening.
+	 * @param endVersion (Optional) End version of event for listening.
 	 * @return List of event updated in Notification Service.
-	 * @throws Exception
 	 */
-	public List<Event> listAllEvents(long startTime) throws Exception {
-		ListAllEventsRequest request = ListAllEventsRequest.newBuilder().setStartTime(startTime).build();
+	public List<EventMeta> listAllEvents(long startTime, long startVersion, long endVersion)
+			throws Exception {
+		ListAllEventsRequest request =
+				ListAllEventsRequest.newBuilder()
+						.setStartTime(startTime)
+						.setStartVersion(startVersion)
+						.setEndVersion(endVersion)
+						.build();
 		ListEventsResponse response = this.notificationServiceStub.listAllEvents(request);
 		return parseEventsFromResponse(response);
 	}
 
-	private static List<Event> parseEventsFromResponse(ListEventsResponse response)
-		throws Exception {
-		if (response.getReturnCode().equals(ReturnStatus.ERROR.toString())) {
-			throw new Exception(response.getReturnMsg());
-		} else {
-			List<Event> result = new ArrayList<>();
+	private static List<EventMeta> parseEventsFromResponse(ListEventsResponse response)
+			throws Exception {
+		if (response.getReturnCode() == ReturnStatus.SUCCESS) {
+			List<EventMeta> eventMetas = new ArrayList<>();
 			for (EventProto eventProto : response.getEventsList()) {
-				result.add(eventProtoToEvent(eventProto));
+				eventMetas.add(buildEventMeta(eventProto));
 			}
-			return result;
+			return eventMetas;
+		} else {
+			throw new Exception(response.getReturnMsg());
 		}
 	}
 
-	/**
-	 *
-	 * @param id The event id after the given id.
-	 * @return List of event updated in Notification Service.
-	 * @throws Exception
-	 */
-	public List<Event> listEventsFromId(long id) throws Exception {
-		ListEventsFromIdRequest request = ListEventsFromIdRequest.newBuilder().setId(id).build();
-		ListEventsResponse response = this.notificationServiceStub.listEventsFromId(request);
-		return parseEventsFromResponse(response);
-	}
-
 	/**
 	 * Start listen specific `key` or `version` notifications in Notification Service.
 	 *
 	 * @param key Key of notification for listening.
 	 * @param watcher Watcher instance for listening notification.
 	 * @param version (Optional) Version of notification for listening.
+	 * @param eventType (Optional) Type of event for listening.
+	 * @param startTime (Optional) Type of event for listening.
 	 */
-	public void startListenEvent(String key, EventWatcher watcher, Integer version) {
-		if (!this.threads.containsKey(key)) {
+	public void startListenEvent(String key, EventWatcher watcher, long version, String eventType, long startTime) {
+		String namespace = this.defaultNamespace;
+		Map<String, String> listenKey = new HashMap<String, String>() {{
+			put(key, namespace);
+		}};
+		if (!this.threads.containsKey(listenKey)) {
+			ArrayList<String> curListenerKeys = new ArrayList<String>(){{add(key);}};
 			EventListener listener = new EventListener(this.notificationServiceStub,
-				key, version, watcher, 5);
+					curListenerKeys, version, eventType, startTime, namespace, watcher, 5);
 			listener.start();
-			this.threads.put(key, listener);
+			this.threads.put(listenKey, listener);
 		}
 	}
 
@@ -166,13 +219,15 @@ public void startListenEvent(String key, EventWatcher watcher, Integer version)
 	 * @param key Key of notification for listening.
 	 */
 	public void stopListenEvent(String key) {
+		String namespace = this.defaultNamespace;
+		Map<String, String> listenKey = new HashMap<String, String>(){{put(key, namespace);}};
 		if (StringUtils.isEmpty(key)) {
-			for (Map.Entry<String, EventListener> entry : threads.entrySet()) {
+			for (Map.Entry<Map<String, String>, EventListener> entry : threads.entrySet()) {
 				entry.getValue().shutdown();
 			}
 		} else {
-			if(this.threads.containsKey(key)) {
-				this.threads.get(key).shutdown();
+			if(this.threads.containsKey(listenKey)) {
+				this.threads.get(listenKey).shutdown();
 			}
 		}
 	}
@@ -182,16 +237,19 @@ public void stopListenEvent(String key) {
 	 *
 	 * @param key Key of notification for listening.
 	 */
-	public long getLatestVersionByKey(String key) throws Exception {
+	public long getLatestVersion(String key) throws Exception {
 		if (StringUtils.isEmpty(key)) {
 			throw new Exception("Empty key, please provide valid key");
 		} else {
-			GetLatestVersionByKeyRequest request = GetLatestVersionByKeyRequest.newBuilder().setKey(key).build();
-			GetLatestVersionResponse response = this.notificationServiceStub.getLatestVersionByKey(request);
+			GetLatestVersionByKeyRequest request =
+					GetLatestVersionByKeyRequest.newBuilder().setKey(key).build();
+			GetLatestVersionResponse response =
+					this.notificationServiceStub.getLatestVersionByKey(request);
 			return parseLatestVersionFromResponse(response);
 		}
 	}
 
+
 	public long parseLatestVersionFromResponse(GetLatestVersionResponse response)
 			throws Exception {
 		if (response.getReturnCode().equals(ReturnStatus.ERROR.toString())) {
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/entity/EventMeta.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/entity/EventMeta.java
new file mode 100644
index 000000000..c03c51a28
--- /dev/null
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/entity/EventMeta.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package com.aiflow.notification.entity;
+
+import com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto;
+
+public class EventMeta {
+
+    private String key;
+    private String value;
+    private String eventType;
+    private long version;
+    private long createTime;
+    private String context;
+    private String namespace;
+
+    public EventMeta(
+            String key,
+            String value,
+            String eventType,
+            long version,
+            long createTime,
+            String context,
+            String namespace) {
+        this.key = key;
+        this.value = value;
+        this.eventType = eventType;
+        this.version = version;
+        this.createTime = createTime;
+        this.context = context;
+        this.namespace = namespace;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    public String getEventType() {
+        return eventType;
+    }
+
+    public void setEventType(String eventType) {
+        this.eventType = eventType;
+    }
+
+    public long getVersion() {
+        return version;
+    }
+
+    public void setVersion(long version) {
+        this.version = version;
+    }
+
+    public long getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(long createTime) {
+        this.createTime = createTime;
+    }
+
+    public String getContext() {
+        return context;
+    }
+
+    public void setContext(String context) {
+        this.context = context;
+    }
+
+    public String getNamespace() {
+        return namespace;
+    }
+
+    public void setNamespace(String namespace) {
+        this.namespace = namespace;
+    }
+
+    @Override
+    public String toString() {
+        return "EventMeta{"
+                + "key='"
+                + key
+                + '\''
+                + ", value='"
+                + value
+                + '\''
+                + ", eventType='"
+                + eventType
+                + '\''
+                + ", version="
+                + version
+                + ", createTime="
+                + createTime
+                + ", context='"
+                + context
+                + '\''
+                + ", namespace='"
+                + namespace
+                + '\''
+                + '}';
+    }
+
+    public static EventMeta buildEventMeta(EventProto eventProto) {
+        return new EventMeta(
+                eventProto.getKey(),
+                eventProto.getValue(),
+                eventProto.getEventType(),
+                eventProto.getVersion(),
+                eventProto.getCreateTime(),
+                eventProto.getContext(),
+                eventProto.getNamespace());
+    }
+}
diff --git a/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2.py b/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2.py
index dad9dc7bb..26c90eca8 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2.py
@@ -20,8 +20,6 @@
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
 # source: notification_service.proto
 
-import sys
-_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1'))
 from google.protobuf.internal import enum_type_wrapper
 from google.protobuf import descriptor as _descriptor
 from google.protobuf import message as _message
@@ -40,8 +38,8 @@
   name='notification_service.proto',
   package='notification_service',
   syntax='proto3',
-  serialized_options=_b('\n\035com.aiflow.notification.proto\210\001\001\220\001\001'),
-  serialized_pb=_b('\n\x1anotification_service.proto\x12\x14notification_service\"\x86\x01\n\nEventProto\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t\x12\x12\n\nevent_type\x18\x03 \x01(\t\x12\x0f\n\x07\x63ontext\x18\x04 \x01(\t\x12\x11\n\tnamespace\x18\x05 \x01(\t\x12\x0f\n\x07version\x18\x06 \x01(\x03\x12\x13\n\x0b\x63reate_time\x18\x07 \x01(\x03\"Q\n\x10SendEventRequest\x12/\n\x05\x65vent\x18\x01 \x01(\x0b\x32 .notification_service.EventProto\x12\x0c\n\x04uuid\x18\x02 \x01(\t\"\x92\x01\n\x12SendEventsResponse\x12/\n\x05\x65vent\x18\x01 \x01(\x0b\x32 .notification_service.EventProto\x12\x37\n\x0breturn_code\x18\x02 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x03 \x01(\t\"\x8c\x01\n\x11ListEventsRequest\x12\x0c\n\x04keys\x18\x01 \x03(\t\x12\x12\n\nevent_type\x18\x02 \x01(\t\x12\x12\n\nstart_time\x18\x03 \x01(\x03\x12\x15\n\rstart_version\x18\x04 \x01(\x03\x12\x17\n\x0ftimeout_seconds\x18\x05 \x01(\x05\x12\x11\n\tnamespace\x18\x06 \x01(\t\"o\n\x14ListAllEventsRequest\x12\x17\n\x0ftimeout_seconds\x18\x01 \x01(\x05\x12\x12\n\nstart_time\x18\x02 \x01(\x03\x12\x15\n\rstart_version\x18\x03 \x01(\x03\x12\x13\n\x0b\x65nd_version\x18\x04 \x01(\x03\"\x93\x01\n\x12ListEventsResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\x12\x30\n\x06\x65vents\x18\x03 \x03(\x0b\x32 .notification_service.EventProto\"W\n\x1cGetLatestVersionByKeyRequest\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\x12\x17\n\x0ftimeout_seconds\x18\x03 \x01(\x05\"T\n\x18GetLatestVersionResponse\x12\x13\n\x0breturn_code\x18\x01 \x01(\t\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\x12\x0f\n\x07version\x18\x03 \x01(\x03\"G\n\x0bMemberProto\x12\x0f\n\x07version\x18\x01 \x01(\x03\x12\x12\n\nserver_uri\x18\x02 \x01(\t\x12\x13\n\x0bupdate_time\x18\x03 \x01(\x03\"(\n\x06Notify\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\"?\n\rNotifyRequest\x12.\n\x08notifies\x18\x01 \x03(\x0b\x32\x1c.notification_service.Notify\"]\n\x0eNotifyResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\"-\n\x12ListMembersRequest\x12\x17\n\x0ftimeout_seconds\x18\x01 \x01(\x05\"\x96\x01\n\x13ListMembersResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\x12\x32\n\x07members\x18\x03 \x03(\x0b\x32!.notification_service.MemberProto\"K\n\x16NotifyNewMemberRequest\x12\x31\n\x06member\x18\x01 \x01(\x0b\x32!.notification_service.MemberProto\"f\n\x17NotifyNewMemberResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t*&\n\x0cReturnStatus\x12\x0b\n\x07SUCCESS\x10\x00\x12\t\n\x05\x45RROR\x10\x01\x32\xf0\x05\n\x13NotificationService\x12_\n\tsendEvent\x12&.notification_service.SendEventRequest\x1a(.notification_service.SendEventsResponse\"\x00\x12\x61\n\nlistEvents\x12\'.notification_service.ListEventsRequest\x1a(.notification_service.ListEventsResponse\"\x00\x12g\n\rlistAllEvents\x12*.notification_service.ListAllEventsRequest\x1a(.notification_service.ListEventsResponse\"\x00\x12U\n\x06notify\x12#.notification_service.NotifyRequest\x1a$.notification_service.NotifyResponse\"\x00\x12\x64\n\x0blistMembers\x12(.notification_service.ListMembersRequest\x1a).notification_service.ListMembersResponse\"\x00\x12p\n\x0fnotifyNewMember\x12,.notification_service.NotifyNewMemberRequest\x1a-.notification_service.NotifyNewMemberResponse\"\x00\x12}\n\x15getLatestVersionByKey\x12\x32.notification_service.GetLatestVersionByKeyRequest\x1a..notification_service.GetLatestVersionResponse\"\x00\x42%\n\x1d\x63om.aiflow.notification.proto\x88\x01\x01\x90\x01\x01\x62\x06proto3')
+  serialized_options=b'\n\035com.aiflow.notification.proto\210\001\001\220\001\001',
+  serialized_pb=b'\n\x1anotification_service.proto\x12\x14notification_service\"\x86\x01\n\nEventProto\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t\x12\x12\n\nevent_type\x18\x03 \x01(\t\x12\x0f\n\x07\x63ontext\x18\x04 \x01(\t\x12\x11\n\tnamespace\x18\x05 \x01(\t\x12\x0f\n\x07version\x18\x06 \x01(\x03\x12\x13\n\x0b\x63reate_time\x18\x07 \x01(\x03\"Q\n\x10SendEventRequest\x12/\n\x05\x65vent\x18\x01 \x01(\x0b\x32 .notification_service.EventProto\x12\x0c\n\x04uuid\x18\x02 \x01(\t\"\x92\x01\n\x12SendEventsResponse\x12/\n\x05\x65vent\x18\x01 \x01(\x0b\x32 .notification_service.EventProto\x12\x37\n\x0breturn_code\x18\x02 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x03 \x01(\t\"\x8c\x01\n\x11ListEventsRequest\x12\x0c\n\x04keys\x18\x01 \x03(\t\x12\x12\n\nevent_type\x18\x02 \x01(\t\x12\x12\n\nstart_time\x18\x03 \x01(\x03\x12\x15\n\rstart_version\x18\x04 \x01(\x03\x12\x17\n\x0ftimeout_seconds\x18\x05 \x01(\x05\x12\x11\n\tnamespace\x18\x06 \x01(\t\"o\n\x14ListAllEventsRequest\x12\x17\n\x0ftimeout_seconds\x18\x01 \x01(\x05\x12\x12\n\nstart_time\x18\x02 \x01(\x03\x12\x15\n\rstart_version\x18\x03 \x01(\x03\x12\x13\n\x0b\x65nd_version\x18\x04 \x01(\x03\"\x93\x01\n\x12ListEventsResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\x12\x30\n\x06\x65vents\x18\x03 \x03(\x0b\x32 .notification_service.EventProto\"W\n\x1cGetLatestVersionByKeyRequest\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\x12\x17\n\x0ftimeout_seconds\x18\x03 \x01(\x05\"T\n\x18GetLatestVersionResponse\x12\x13\n\x0breturn_code\x18\x01 \x01(\t\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\x12\x0f\n\x07version\x18\x03 \x01(\x03\"G\n\x0bMemberProto\x12\x0f\n\x07version\x18\x01 \x01(\x03\x12\x12\n\nserver_uri\x18\x02 \x01(\t\x12\x13\n\x0bupdate_time\x18\x03 \x01(\x03\"(\n\x06Notify\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\"?\n\rNotifyRequest\x12.\n\x08notifies\x18\x01 \x03(\x0b\x32\x1c.notification_service.Notify\"]\n\x0eNotifyResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\"-\n\x12ListMembersRequest\x12\x17\n\x0ftimeout_seconds\x18\x01 \x01(\x05\"\x96\x01\n\x13ListMembersResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t\x12\x32\n\x07members\x18\x03 \x03(\x0b\x32!.notification_service.MemberProto\"K\n\x16NotifyNewMemberRequest\x12\x31\n\x06member\x18\x01 \x01(\x0b\x32!.notification_service.MemberProto\"f\n\x17NotifyNewMemberResponse\x12\x37\n\x0breturn_code\x18\x01 \x01(\x0e\x32\".notification_service.ReturnStatus\x12\x12\n\nreturn_msg\x18\x02 \x01(\t*&\n\x0cReturnStatus\x12\x0b\n\x07SUCCESS\x10\x00\x12\t\n\x05\x45RROR\x10\x01\x32\xf0\x05\n\x13NotificationService\x12_\n\tsendEvent\x12&.notification_service.SendEventRequest\x1a(.notification_service.SendEventsResponse\"\x00\x12\x61\n\nlistEvents\x12\'.notification_service.ListEventsRequest\x1a(.notification_service.ListEventsResponse\"\x00\x12g\n\rlistAllEvents\x12*.notification_service.ListAllEventsRequest\x1a(.notification_service.ListEventsResponse\"\x00\x12U\n\x06notify\x12#.notification_service.NotifyRequest\x1a$.notification_service.NotifyResponse\"\x00\x12\x64\n\x0blistMembers\x12(.notification_service.ListMembersRequest\x1a).notification_service.ListMembersResponse\"\x00\x12p\n\x0fnotifyNewMember\x12,.notification_service.NotifyNewMemberRequest\x1a-.notification_service.NotifyNewMemberResponse\"\x00\x12}\n\x15getLatestVersionByKey\x12\x32.notification_service.GetLatestVersionByKeyRequest\x1a..notification_service.GetLatestVersionResponse\"\x00\x42%\n\x1d\x63om.aiflow.notification.proto\x88\x01\x01\x90\x01\x01\x62\x06proto3'
 )
 
 _RETURNSTATUS = _descriptor.EnumDescriptor(
@@ -82,35 +80,35 @@
     _descriptor.FieldDescriptor(
       name='key', full_name='notification_service.EventProto.key', index=0,
       number=1, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='value', full_name='notification_service.EventProto.value', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='event_type', full_name='notification_service.EventProto.event_type', index=2,
       number=3, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='context', full_name='notification_service.EventProto.context', index=3,
       number=4, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='namespace', full_name='notification_service.EventProto.namespace', index=4,
       number=5, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -162,7 +160,7 @@
     _descriptor.FieldDescriptor(
       name='uuid', full_name='notification_service.SendEventRequest.uuid', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -207,7 +205,7 @@
     _descriptor.FieldDescriptor(
       name='return_msg', full_name='notification_service.SendEventsResponse.return_msg', index=2,
       number=3, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -245,7 +243,7 @@
     _descriptor.FieldDescriptor(
       name='event_type', full_name='notification_service.ListEventsRequest.event_type', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -273,7 +271,7 @@
     _descriptor.FieldDescriptor(
       name='namespace', full_name='notification_service.ListEventsRequest.namespace', index=5,
       number=6, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -363,7 +361,7 @@
     _descriptor.FieldDescriptor(
       name='return_msg', full_name='notification_service.ListEventsResponse.return_msg', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -401,14 +399,14 @@
     _descriptor.FieldDescriptor(
       name='key', full_name='notification_service.GetLatestVersionByKeyRequest.key', index=0,
       number=1, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='namespace', full_name='notification_service.GetLatestVersionByKeyRequest.namespace', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -446,14 +444,14 @@
     _descriptor.FieldDescriptor(
       name='return_code', full_name='notification_service.GetLatestVersionResponse.return_code', index=0,
       number=1, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='return_msg', full_name='notification_service.GetLatestVersionResponse.return_msg', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -498,7 +496,7 @@
     _descriptor.FieldDescriptor(
       name='server_uri', full_name='notification_service.MemberProto.server_uri', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -536,14 +534,14 @@
     _descriptor.FieldDescriptor(
       name='key', full_name='notification_service.Notify.key', index=0,
       number=1, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='namespace', full_name='notification_service.Notify.namespace', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -612,7 +610,7 @@
     _descriptor.FieldDescriptor(
       name='return_msg', full_name='notification_service.NotifyResponse.return_msg', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -681,7 +679,7 @@
     _descriptor.FieldDescriptor(
       name='return_msg', full_name='notification_service.ListMembersResponse.return_msg', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),
@@ -757,7 +755,7 @@
     _descriptor.FieldDescriptor(
       name='return_msg', full_name='notification_service.NotifyNewMemberResponse.return_msg', index=1,
       number=2, type=9, cpp_type=9, label=1,
-      has_default_value=False, default_value=_b("").decode('utf-8'),
+      has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR),


Test patch:
diff --git a/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java b/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java
index 4689bbeb6..543fdb06d 100644
--- a/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java
+++ b/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java
@@ -18,6 +18,7 @@
  */
 package com.aiflow.notification.client;
 
+import com.aiflow.notification.entity.EventMeta;
 import com.aiflow.notification.service.LocalNotificationService;
 import io.grpc.ManagedChannel;
 import io.grpc.inprocess.InProcessChannelBuilder;
@@ -29,7 +30,10 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import static org.junit.Assert.*;
 
@@ -48,23 +52,27 @@ public void setUp() throws IOException {
 
 		// Create a server, add service, start, and register for automatic graceful shutdown.
 		this.grpcCleanup.register(InProcessServerBuilder
-			.forName(serverName).directExecutor().fallbackHandlerRegistry(this.serviceRegistry)
-			.addService(mockNotificationService).build().start());
+				.forName(serverName).directExecutor().fallbackHandlerRegistry(this.serviceRegistry)
+				.addService(mockNotificationService).build().start());
 
 		// Create a client channel and register for automatic graceful shutdown.
 		ManagedChannel channel = this.grpcCleanup.register(
-			InProcessChannelBuilder.forName(serverName).directExecutor().build());
+				InProcessChannelBuilder.forName(serverName).directExecutor().build());
 
 		// Create a NotificationClient using the in-process channel
-		this.client = new NotificationClient(channel);
+		this.client = new NotificationClient("localhost:50051", "default", false,
+				5, 10, 2000);
 	}
 
 	@Test
 	public void sendEvent() throws Exception {
+		long latestVersion = this.client.getLatestVersion("key");
 		for(int i = 0; i <3; i++) {
-			Event event =this.client.sendEvent("key", String.valueOf(i), "type");
+			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
 		}
-		List<Event> eventList = this.client.listEvents("key", 0);
+		ArrayList<String> listenerKeys = new ArrayList<String>(){{add("key");}};
+		long startTime = System.currentTimeMillis();
+		List<EventMeta> eventList = this.client.listEvents(listenerKeys, latestVersion, "type", 0);
 		assertEquals(3, eventList.size());
 	}
 
@@ -72,51 +80,44 @@ public void sendEvent() throws Exception {
 	public void listAllEvents() throws Exception{
 		long startTime=0;
 		for(int i = 0; i <3; i++) {
-			Event event =this.client.sendEvent("key", String.valueOf(i), "type");
+			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
 			if(i == 1){
-				startTime = event.getCreate_time();
+				startTime = event.getCreateTime();
 			}
 		}
-		List<Event> eventList = this.client.listAllEvents(startTime);
-		assertEquals(2, eventList.size());
-	}
-
-	@Test
-	public void listEventsFromId() throws Exception {
-		for(int i = 0; i <3; i++) {
-			Event event =this.client.sendEvent("key", String.valueOf(i), "type");
-		}
-		List<Event> eventList = this.client.listEventsFromId(1l);
+		List<EventMeta> eventList = this.client.listAllEvents(startTime, 0, 0);
 		assertEquals(2, eventList.size());
 	}
 
 	@Test
 	public void startListenEvent() throws Exception{
+		long latestVersion = this.client.getLatestVersion("key");
 		for(int i = 0; i <3; i++) {
-			Event event =this.client.sendEvent("key", String.valueOf(i), "type");
+			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
 
 		}
 		final Integer[] ii = { 0 };
-		this.client.startListenEvent("key", new EventWatcher() {
+		String listenerKey = "key";
+		this.client.startListenEvent(listenerKey, new EventWatcher() {
 			@Override
-			public void process(List<Event> events) {
+			public void process(List<EventMeta> events) {
 				ii[0] += events.size();
 			}
-		}, 0);
+		}, latestVersion, "type", 0);
 
-		Thread.sleep(1000);
+		Thread.sleep(10000);
 		this.client.stopListenEvent("key");
 		assertEquals(3, ii[0].intValue());
 	}
 
 	@Test
 	public void getLatestVersion() throws Exception {
+		long latestVersion = this.client.getLatestVersion("key");
 		for(int i = 0; i <3; i++) {
-			Event event =this.client.sendEvent("key", String.valueOf(i), "type");
+			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
 
 		}
-		long latestVersion = this.client.getLatestVersionByKey("key");
-		assertEquals(3, latestVersion);
+		long newLatestVersion = this.client.getLatestVersion("key");
+		assertEquals(latestVersion + 3, newLatestVersion);
 	}
-
 }
diff --git a/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/service/LocalNotificationService.java b/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/service/LocalNotificationService.java
index a48911d9c..6449efcd5 100644
--- a/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/service/LocalNotificationService.java
+++ b/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/service/LocalNotificationService.java
@@ -49,13 +49,12 @@ public void sendEvent(NotificationServiceOuterClass.SendEventRequest request,
 			.setValue(request.getEvent().getValue())
 			.setEventType(request.getEvent().getEventType())
 			.setCreateTime(System.nanoTime())
-			.setVersion(version.intValue())
-			.setId(id.intValue()).build();
+			.setVersion(version.intValue()).build();
 
 		store.add(eventProto);
 		NotificationServiceOuterClass.SendEventsResponse response
 			= NotificationServiceOuterClass.SendEventsResponse.newBuilder().setEvent(eventProto).setReturnCode(
-			String.valueOf(NotificationServiceOuterClass.ReturnStatus.SUCCESS)).build();
+			NotificationServiceOuterClass.ReturnStatus.SUCCESS).build();
 		responseObserver.onNext(response);
 		responseObserver.onCompleted();
 	}
@@ -63,8 +62,8 @@ public void sendEvent(NotificationServiceOuterClass.SendEventRequest request,
 	@Override
 	public void listEvents(NotificationServiceOuterClass.ListEventsRequest request,
 		StreamObserver<NotificationServiceOuterClass.ListEventsResponse> responseObserver) {
-		String key = request.getEvent().getKey();
-		int version = request.getEvent().getVersion();
+		String key = request.getKeys(0);
+		long version = request.getStartVersion();
 		List<NotificationServiceOuterClass.EventProto> eventProtos = new ArrayList<>();
 		for (NotificationServiceOuterClass.EventProto eventProto : store) {
 			if (key.equals(eventProto.getKey()) && eventProto.getVersion() > version) {
@@ -101,22 +100,6 @@ public void listAllEvents(NotificationServiceOuterClass.ListAllEventsRequest req
 		responseObserver.onCompleted();
 	}
 
-	@Override
-	public void listEventsFromId(NotificationServiceOuterClass.ListEventsFromIdRequest request,
-		StreamObserver<NotificationServiceOuterClass.ListEventsResponse> responseObserver) {
-		List<NotificationServiceOuterClass.EventProto> eventProtos = new ArrayList<>();
-		for (NotificationServiceOuterClass.EventProto eventProto : store) {
-			if (eventProto.getId() > request.getId()) {
-				eventProtos.add(eventProto);
-			}
-		}
-		NotificationServiceOuterClass.ListEventsResponse response
-			= NotificationServiceOuterClass.ListEventsResponse.newBuilder()
-			.addAllEvents(eventProtos).build();
-		responseObserver.onNext(response);
-		responseObserver.onCompleted();
-	}
-
 	@Override
 	public void getLatestVersionByKey(NotificationServiceOuterClass.GetLatestVersionByKeyRequest request,
 									  StreamObserver<NotificationServiceOuterClass.GetLatestVersionResponse> responseObserver) {