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

Issue:
frontend subscribe events



Discussion / hints:

Fix patch:
diff --git a/flink-ai-flow/ai_flow/airflow/dag_generator.py b/flink-ai-flow/ai_flow/airflow/dag_generator.py
index 3ff20e1f1..2d94556b4 100644
--- a/flink-ai-flow/ai_flow/airflow/dag_generator.py
+++ b/flink-ai-flow/ai_flow/airflow/dag_generator.py
@@ -16,15 +16,15 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-from datetime import datetime
 from abc import ABC
+from datetime import datetime
 from typing import Text, List
 
-from ai_flow.plugins.engine import DummyEngine
-from ai_flow.workflow.job_config import PeriodicConfig
-from ai_flow.common.registry import BaseRegistry
 from ai_flow.common import json_utils
+from ai_flow.common.registry import BaseRegistry
 from ai_flow.graph.edge import MetConfig, MetCondition, EventLife, MetValueCondition, TaskAction
+from ai_flow.plugins.engine import DummyEngine
+from ai_flow.workflow.job_config import PeriodicConfig
 from ai_flow.workflow.workflow import Workflow
 
 
@@ -47,7 +47,7 @@ def job_name_to_task_id(job_name):
     elif first_index == last_index:
         return job_name
     else:
-        return job_name[first_index+1: last_index]
+        return job_name[first_index + 1: last_index]
 
 
 class AirflowCodeGenerator(ABC):
@@ -88,9 +88,9 @@ class DAGTemplate(object):
 import datetime
 from airflow.contrib.jobs.event_handlers import AIFlowHandler
 from airflow.operators.bash import BashOperator\n"""
-# from ai_flow.deployer.utils.kubernetes_util import load_kubernetes_config
-# import ai_flow as af
-# load_kubernetes_config()\n"""
+    # from ai_flow.deployer.utils.kubernetes_util import load_kubernetes_config
+    # import ai_flow as af
+    # load_kubernetes_config()\n"""
 
     SET_CONFIG = """af.set_project_config_file('{0}')\naf.set_master_config()\n"""
 
@@ -100,11 +100,11 @@ class DAGTemplate(object):
     DAG_DEFINE = """dag = DAG(dag_id='{0}', default_args=default_args)\n"""
 
     PERIODIC_CONFIG = """op_{0}.executor_config = {{'periodic_config': {1}}}
-op_{0}.subscribe_event('UNREACHED_EVENT', 'UNREACHED_EVENT', 'UNREACHED_EVENT')\n"""
+op_{0}.subscribe_event('UNREACHED_EVENT', 'UNREACHED_EVENT', 'UNREACHED_EVENT', None)\n"""
 
     UPSTREAM_OP = """{0}.set_upstream({1})\n"""
 
-    EVENT_DEPS = """{0}.subscribe_event('{1}', '{2}', '{3}')\n"""
+    EVENT_DEPS = """{0}.subscribe_event('{1}', '{2}', '{3}', '{4}')\n"""
 
     MET_HANDLER = """configs_{0}='{1}'\n
 {0}.set_events_handler(AIFlowHandler(configs_{0}))\n"""
@@ -118,13 +118,14 @@ def generate_op_code(self, job):
         self.op_count += 1
         generator: AirflowCodeGenerator = get_airflow_code_manager().get_generator(job.platform, job.exec_engine)
         code_text = generator.generate_code(op_index=self.op_count, job=job)
-        return job.instance_id, "op_{0}".format(self.op_count), code_text
+        return job.instance_id, "op_{0}".format(self.op_count), code_text, job.job_name
 
     def generate_upstream(self, op_1, op_2):
         return DAGTemplate.UPSTREAM_OP.format(op_1, op_2)
 
-    def generate_event_deps(self, op, met_config):
-        return DAGTemplate.EVENT_DEPS.format(op, met_config.event_key, met_config.event_type, met_config.namespace)
+    def generate_event_deps(self, op, from_task_id, met_config):
+        return DAGTemplate.EVENT_DEPS.format(op, met_config.event_key, met_config.event_type,
+                                             met_config.namespace, from_task_id)
 
     def generate_handler(self, op, configs: List[MetConfig]):
         return DAGTemplate.MET_HANDLER.format(op, json_utils.dumps(configs))
@@ -144,7 +145,8 @@ def generator(self,
         op_set = set()
         for name, job in workflow.jobs.items():
             if job.exec_engine != DummyEngine.engine():
-                generator: AirflowCodeGenerator = get_airflow_code_manager().get_generator(job.platform, job.exec_engine)
+                generator: AirflowCodeGenerator = get_airflow_code_manager().get_generator(job.platform,
+                                                                                           job.exec_engine)
                 if generator not in op_set:
                     code_text += generator.generate_operator_code()
                     op_set.add(generator)
@@ -154,10 +156,12 @@ def generator(self,
         code_text += DAGTemplate.DAG_DEFINE.format(dag_id)
 
         task_map = {}
+        job_name_map = {}
         for name, job in workflow.jobs.items():
             if job.exec_engine != DummyEngine.engine():
-                task_id, op_name, code = self.generate_op_code(job)
+                task_id, op_name, code, job_name = self.generate_op_code(job)
                 task_map[task_id] = op_name
+                job_name_map[task_id] = job_name
                 code_text += code
                 # add periodic
                 if job.job_config.periodic_config is not None:
@@ -183,7 +187,11 @@ def generator(self,
                         code = self.generate_upstream(op_name, task_map[dep_task_id])
                         code_text += code
                     else:
-                        code = self.generate_event_deps(op_name, met_config)
+                        if edge.target_node_id in job_name_map:
+                            from_op_name = job_name_map[edge.target_node_id]
+                        else:
+                            from_op_name = ''
+                        code = self.generate_event_deps(op_name, job_name_to_task_id(from_op_name), met_config)
                         code_text += code
                         configs.append(met_config)
                 if len(configs) > 0:
diff --git a/flink-ai-flow/flink_ai_flow/local_flink_job.py b/flink-ai-flow/flink_ai_flow/local_flink_job.py
index 85134932c..6a945ad8c 100644
--- a/flink-ai-flow/flink_ai_flow/local_flink_job.py
+++ b/flink-ai-flow/flink_ai_flow/local_flink_job.py
@@ -320,9 +320,9 @@ def generate_code(self, op_index, job: LocalFlinkJob):
                 ' '.join(job.exec_cmd),
                 sys_env)
         else:
-            return """from flink_ai_flow.local_flink_job import LocalFlinkOperator
+            return """from flink_ai_flow.local_flink_job import FlinkOperator
 env_{0}={4}
-op_{0} = LocalFlinkOperator(task_id='{1}', bash_command='{2}', properties='{3}', dag=dag, env=env_{0})\n""".format(
+op_{0} = FlinkOperator(task_id='{1}', bash_command='{2}', properties='{3}', dag=dag, env=env_{0})\n""".format(
                 op_index,
                 job_name_to_task_id(job.job_name),
                 ' '.join(job.exec_cmd),
@@ -347,13 +347,15 @@ def stop_job(self, job: AbstractJob):
                 os.remove(job_execution_path)
 
 
-class LocalFlinkOperator(BashOperator):
+class FlinkOperator(BashOperator):
+
+    ui_color = '#d9ff05'
 
     def __init__(
             self,
             properties,
             *args, **kwargs):
-        super(LocalFlinkOperator, self).__init__(*args, **kwargs)
+        super(FlinkOperator, self).__init__(*args, **kwargs)
         self.properties = json.loads(properties)
 
     def on_kill(self):
diff --git a/flink-ai-flow/lib/airflow/airflow/models/baseoperator.py b/flink-ai-flow/lib/airflow/airflow/models/baseoperator.py
index 25c3ca947..088fb044b 100644
--- a/flink-ai-flow/lib/airflow/airflow/models/baseoperator.py
+++ b/flink-ai-flow/lib/airflow/airflow/models/baseoperator.py
@@ -45,7 +45,6 @@
 import jinja2
 from cached_property import cached_property
 from dateutil.relativedelta import relativedelta
-from notification_service.base_notification import UNDEFINED_EVENT_TYPE, BaseEvent
 from sqlalchemy.orm import Session
 
 from airflow.configuration import conf
@@ -72,6 +71,7 @@
 from airflow.utils.session import provide_session
 from airflow.utils.trigger_rule import TriggerRule
 from airflow.utils.weight_rule import WeightRule
+from notification_service.base_notification import UNDEFINED_EVENT_TYPE, BaseEvent
 
 if TYPE_CHECKING:
     from airflow.utils.task_group import TaskGroup  # pylint: disable=cyclic-import
@@ -137,10 +137,13 @@ def __init__(self, event_handler: EventHandler = None):
         :type event_handler: EventHandler
         """
         super().__init__()
-        self._subscribed_events: Set[Tuple[str, str, str]] = set()
+        self._subscribed_events: Set[Tuple[str, str, str, str]] = set()
         self._events_handler: EventHandler = event_handler
 
-    def subscribe_event(self, event_key: str, event_type: str = UNDEFINED_EVENT_TYPE, event_namespace: str = 'default'):
+    def subscribe_event(self, event_key: str,
+                        event_type: str = UNDEFINED_EVENT_TYPE,
+                        event_namespace: str = 'default',
+                        from_task_id: str = None):
         """
         Subscribe to the events with the event_key and event_type. The event_handler will only handle the
         event that the operator subscribes to. event_type is optional, if it is not specify, it subscribes to all type
@@ -151,14 +154,16 @@ def subscribe_event(self, event_key: str, event_type: str = UNDEFINED_EVENT_TYPE
         :type event_namespace: str
         :param event_type: the type of the event to subscribe to.
         :type event_type: str
+        :param from_task_id: which task send the event
+        :type from_task_id: str
         :return: None
         """
-        self._subscribed_events.add((event_namespace, event_key, event_type))
+        self._subscribed_events.add((event_namespace, event_key, event_type, from_task_id))
 
-    def get_subscribed_events(self) -> Set[Tuple[str, str, str]]:
+    def get_subscribed_events(self) -> Set[Tuple[str, str, str, str]]:
         """
         :return: the set of events that the operator subscribes to.
-        :rtype: Set[str, str, str]
+        :rtype: Set[str, str, str, str]
         """
         return self._subscribed_events
 
diff --git a/flink-ai-flow/lib/airflow/airflow/models/dag.py b/flink-ai-flow/lib/airflow/airflow/models/dag.py
index dd864def8..d27c1ac1e 100644
--- a/flink-ai-flow/lib/airflow/airflow/models/dag.py
+++ b/flink-ai-flow/lib/airflow/airflow/models/dag.py
@@ -2044,15 +2044,15 @@ def __init__(self, dag: DAG = None):
         # value: key: event_key
         #        value:
         #           key: event_type
-        #           value: dict(task_id: None)
+        #           value: dict(task_id: from_task_id)
         self.task_event_dependencies = {}
         if dag is not None:
             for task_id, op in dag.task_dict.items():
-                dep_set: Set[Tuple[str, str, str]] = op.get_subscribed_events()
-                for event_namespace, event_key, event_type in dep_set:
-                    self.add_dependencies(task_id, EventKey(event_key, event_type, event_namespace))
+                dep_set: Set[Tuple[str, str, str, str]] = op.get_subscribed_events()
+                for event_namespace, event_key, event_type, from_task_id in dep_set:
+                    self.add_dependencies(task_id, EventKey(event_key, event_type, event_namespace), from_task_id)
 
-    def add_dependencies(self, task_id: str, event_key: EventKey):
+    def add_dependencies(self, task_id: str, event_key: EventKey, from_task_id: str = None):
         if event_key.namespace not in self.task_event_dependencies:
             self.task_event_dependencies[event_key.namespace] = {}
         namespace_dict = self.task_event_dependencies[event_key.namespace]
@@ -2061,7 +2061,7 @@ def add_dependencies(self, task_id: str, event_key: EventKey):
         key_dict = namespace_dict[event_key.key]
         if event_key.event_type not in key_dict:
             key_dict[event_key.event_type] = {}
-        key_dict[event_key.event_type][task_id] = None
+        key_dict[event_key.event_type][task_id] = from_task_id
 
     def find_affected_tasks(self, event_key: EventKey)->Optional[List]:
         result = set()
diff --git a/flink-ai-flow/lib/airflow/airflow/www/static/js/events.js b/flink-ai-flow/lib/airflow/airflow/www/static/js/events.js
new file mode 100644
index 000000000..78536159d
--- /dev/null
+++ b/flink-ai-flow/lib/airflow/airflow/www/static/js/events.js
@@ -0,0 +1,34 @@
+/*!
+ * 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.
+ */
+
+/* global window, dagTZ, moment, convertSecsToHumanReadable */
+
+// We don't re-import moment again, otherwise webpack will include it twice in the bundle!
+import { escapeHtml } from './main';
+
+export default function eTooltip(e, { includeTryNumber = false } = {}) {
+  let tt = '';
+  tt += `<strong>Namespace:</strong> ${escapeHtml(e[0])}<br><br>`;
+  tt += `Event_Key: ${escapeHtml(e[1])}<br>`;
+  tt += `Event_Type: ${escapeHtml(e[2])}<br><br>`;
+  tt +=`<em><strong>Send_Count:</strong> ${escapeHtml(e[3])}</em>`;
+  return tt;
+}
+
+window.eTooltip = eTooltip;
diff --git a/flink-ai-flow/lib/airflow/airflow/www/templates/airflow/graph.html b/flink-ai-flow/lib/airflow/airflow/www/templates/airflow/graph.html
index a23484bb3..730ba3c97 100644
--- a/flink-ai-flow/lib/airflow/airflow/www/templates/airflow/graph.html
+++ b/flink-ai-flow/lib/airflow/airflow/www/templates/airflow/graph.html
@@ -31,6 +31,9 @@
         stroke: {{state_color}};
       }
     {% endfor %}
+      g.node.event_sent rect {
+        stroke: #7b09f7;
+      }
   </style>
 {% endblock %}
 
@@ -82,12 +85,12 @@
   <div class="legend-row">
     <div>
       {% for op in operators %}<span class="legend-item" style="color: {{ op.ui_fgcolor }};background: {{ op.ui_color }};">
-        {{ op.task_type }}</span>{% endfor %}
-      {% if events != '{}' %}<span class="legend-item" style="color: #000;background: #e1dcf5;">event</span>{% endif %}
+        {{ op.task_type }}</span>{% endfor %}{% if events != {} %}<span class="legend-item" style="color: #000;background: #ccb4fb;">event</span>{% endif %}
     </div>
     <div>
       {% for state, state_color in state_color_mapping.items() %}<span class="legend-item js-state-legend-item" data-state="{{state}}" style="border-color: {{state_color}};">
-        {{state}}</span>{% endfor %}<span class="legend-item legend-item--no-border js-state-legend-item" data-state="no_status" style="border-color:white;">no_status</span>
+        {{state}}</span>{% endfor %}{% if events != {} %}<span class="legend-item js-state-legend-item" data-state="event_sent" style="border-color: #7b09f7;">event_sent</span>{% endif %}
+      <span class="legend-item legend-item--no-border js-state-legend-item" data-state="no_status" style="border-color: white;">no_status</span>
     </div>
   </div>
   <div id="error" style="display: none; margin-top: 10px;" class="alert alert-danger" role="alert">
@@ -124,6 +127,7 @@
   <script src="{{ url_for_asset('dagre-d3.min.js') }}"></script>
   <script src="{{ url_for_asset('d3-tip.js') }}"></script>
   <script src="{{ url_for_asset('taskInstances.js') }}"></script>
+  <script src="{{ url_for_asset('events.js') }}"></script>
   <script>
 
       var highlight_color = "#000";
@@ -144,7 +148,8 @@
       // Below variables are being used in dag.js
       var tasks = {{ tasks|tojson }};
       var task_instances = {{ task_instances|tojson }};
-      var getTaskInstanceURL = "{{ url_for('Airflow.task_instances') }}" +
+      var events = {{ events|tojson }};
+      var getNodeURL = "{{ url_for('Airflow.nodes') }}" +
         "?dag_id=" + encodeURIComponent(dag_id) + "&execution_date=" +
         encodeURIComponent(execution_date);
 
@@ -246,7 +251,7 @@
           d3.selectAll("g.edgePath")
             .style("opacity", 1);
         });
-        updateNodesStates(task_instances);
+        updateNodesStates(task_instances, events);
         setUpZoomSupport();
       }
 
@@ -336,7 +341,7 @@
           var children = get_children_ids(node);
           for(const child of children) {
             if(child.indexOf(search_text) > -1)
-              return true
+              return true;
           }
         }
       }
@@ -450,19 +455,19 @@
       function stateIsSet(){
           for (var key in stateFocusMap){
               if (stateFocusMap[key]){
-                  return true
+                  return true;
               }
           }
-          return false
+          return false;
       }
 
       function handleRefresh() {
         $('#loading-dots').css('display', 'inline-block');
-        $.get(getTaskInstanceURL)
+        $.get(getNodeURL)
           .done(
-            (tis) => {
-              task_instances = JSON.parse(tis)
-              updateNodesStates(task_instances);
+            (ns) => {
+              nodes = JSON.parse(ns)
+              updateNodesStates(nodes.task_instances, nodes.events);
               setTimeout(function() { $('#loading-dots').hide(); }, 500);
               $('#error').hide();
             }
@@ -548,20 +553,22 @@
 
       // Assigning css classes based on state to nodes
       // Initiating the tooltips
-      function updateNodesStates(tis) {
+      function updateNodesStates(tis, es) {
         for(const node_id of g.nodes())
         {
           elem = g.node(node_id).elem;
-          elem.setAttribute("class", "node enter " + get_node_state(node_id, tis));
+          elem.setAttribute("class", "node enter " + get_node_state(node_id, tis, es));
           elem.setAttribute("data-toggle", "tooltip");
 
-          const task_id = node_id;
           elem.onmouseover = (evt) => {
-            if(task_id in tis) {
-              const tt = tiTooltip(tis[task_id]);
+            if(node_id in tis) {
+              const tt = tiTooltip(tis[node_id]);
               taskTip.show(tt, evt.target); // taskTip is defined in graph.html
-            } else if(task_group_tips.has(task_id)) {
-              const tt = group_tooltip(task_id, tis)
+            } else if (node_id in es) {
+              const tt = eTooltip(es[node_id]);
+              taskTip.show(tt, evt.target);
+            } else if (task_group_tips.has(node_id)) {
+              const tt = group_tooltip(node_id, tis);
               taskTip.show(tt, evt.target);
             }
           };
@@ -575,7 +582,7 @@
       function get_children_ids(group) {
         var children = []
         for(const [key, val] of Object.entries(group.children)) {
-          if(val.children == undefined) {
+          if (val.children == undefined) {
             // node
             children.push(val.id)
           } else {
@@ -586,28 +593,31 @@
             }
           }
         }
-        return children
+        return children;
       }
 
 
       // Return the state for the node based on the state of its taskinstance or that of its
       // children if it's a group node
-      function get_node_state(node_id, tis) {
-        node = g.node(node_id)
+      function get_node_state(node_id, tis, es) {
+        if (node_id in es && es[node_id][3] > 0)
+          return 'event_sent';
+
+        node = g.node(node_id);
 
         if (node.children == undefined) {
           if(node_id in tis)
-            return tis[node_id].state
+            return tis[node_id].state;
 
-          return "no_status"
+          return "no_status";
         }
-        var children = get_children_ids(node)
+        var children = get_children_ids(node);
 
-        children_states = new Set()
+        children_states = new Set();
         children.forEach(function(task_id) {
           if (task_id in tis) {
-            var state = tis[task_id].state
-            children_states.add(state == null ? "no_status" : state)
+            var state = tis[task_id].state;
+            children_states.add(state == null ? "no_status" : state);
           }
         })
 
@@ -615,13 +625,13 @@
         // the group state.
         var priority = ["failed", "upstream_failed", "up_for_retry","up_for_reschedule",
                         "queued", "scheduled", "sensing", "running", "shutdown", "removed",
-                        "no_status", "success", "skipped"]
+                        "no_status", "success", "skipped"];
 
         for(const state of priority) {
           if (children_states.has(state))
-            return state
+            return state;
         }
-        return "no_status"
+        return "no_status";
       }
 
       // Focus the graph on the expanded/collapsed node
diff --git a/flink-ai-flow/lib/airflow/airflow/www/views.py b/flink-ai-flow/lib/airflow/airflow/www/views.py
index a700b8748..150c26ad9 100644
--- a/flink-ai-flow/lib/airflow/airflow/www/views.py
+++ b/flink-ai-flow/lib/airflow/airflow/www/views.py
@@ -28,7 +28,7 @@
 from datetime import datetime, timedelta
 from json import JSONDecodeError
 from operator import itemgetter
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Set, Tuple
 from urllib.parse import parse_qsl, unquote, urlencode, urlparse
 
 import lazy_object_proxy
@@ -88,7 +88,7 @@
 from airflow.security import permissions
 from airflow.serialization.serialized_objects import BaseSerialization
 from airflow.ti_deps.dep_context import DepContext
-from airflow.ti_deps.dependencies_deps import RUNNING_DEPS, SCHEDULER_QUEUED_DEPS
+from airflow.ti_deps.dependencies_deps import SCHEDULER_QUEUED_DEPS
 from airflow.utils import json as utils_json, timezone
 from airflow.utils.dates import infer_time_unit, scale_time_units
 from airflow.utils.helpers import alchemy_to_dict
@@ -109,6 +109,7 @@
     TaskInstanceEditForm,
 )
 from airflow.www.widgets import AirflowModelListWidget
+from notification_service.client import NotificationClient
 from notification_service.util.db import EventModel
 
 PAGE_SIZE = conf.getint('webserver', 'page_size')
@@ -185,7 +186,7 @@ def get_date_time_num_runs_dag_runs_form_data(www_request, session, dag):
     }
 
 
-def task_group_to_dict(task_group, events):
+def task_group_to_dict(task_group):
     """
     Create a nested dict representation of this TaskGroup and its children used to construct
     the Graph View.
@@ -194,21 +195,18 @@ def task_group_to_dict(task_group, events):
         nodes = []
         subscribed_events = task_group.get_subscribed_events()
         if subscribed_events:
-            dependencies = []
-            for event_namespace, event_key, event_type in BaseSerialization._deserialize(subscribed_events):
+            for event_namespace, event_key, event_type, from_task_id in BaseSerialization._deserialize(subscribed_events):
                 event = '{},{},{}'.format(event_namespace, event_key, event_type)
-                dependencies.append([event_namespace, event_key, event_type])
                 nodes.append({
                     'id': event,
                     'value': {
                         'label': event,
                         'labelStyle': 'fill:#000;',
-                        'style': 'fill:#e1dcf5;',
+                        'style': 'fill:#ccb4fb;',
                         'rx': 5,
                         'ry': 5,
                     }
                 })
-            events.update({task_group.task_id: dependencies})
 
         nodes.append({
             'id': task_group.task_id,
@@ -224,7 +222,7 @@ def task_group_to_dict(task_group, events):
 
     children = []
     for child in sorted(task_group.children.values(), key=lambda t: t.label):
-        children.extend(task_group_to_dict(child, events))
+        children.extend(task_group_to_dict(child))
 
     if task_group.upstream_group_ids or task_group.upstream_task_ids:
         children.append(
@@ -357,10 +355,14 @@ def get_downstream(task):
                 get_downstream(child)
         subscribed_events = task.get_subscribed_events()
         if subscribed_events:
-            for event_namespace, event_key, event_type in BaseSerialization._deserialize(subscribed_events):
-                edge = ('{},{},{}'.format(event_namespace, event_key, event_type), task.task_id)
-                if edge not in edges:
-                    edges.add(edge)
+            for event_namespace, event_key, event_type, from_task_id in BaseSerialization._deserialize(subscribed_events):
+                to_edge = ('{},{},{}'.format(event_namespace, event_key, event_type), task.task_id)
+                if to_edge not in edges:
+                    edges.add(to_edge)
+                if from_task_id:
+                    from_edge = (from_task_id, '{},{},{}'.format(event_namespace, event_key, event_type))
+                    if from_edge not in edges:
+                        edges.add(from_edge)
 
     for root in dag.roots:
         get_downstream(root)
@@ -439,8 +441,8 @@ class Airflow(AirflowBaseView):  # noqa: D101  pylint: disable=too-many-public-m
     def __init__(self, server_uri=None, **kwargs):
         super().__init__(**kwargs)
         if server_uri:
-            self.scheduler_client: EventSchedulerClient = EventSchedulerClient(server_uri=server_uri,
-                                                                               namespace=SCHEDULER_NAMESPACE)
+            self.notification_client: NotificationClient = NotificationClient(server_uri, SCHEDULER_NAMESPACE)
+            self.scheduler_client: EventSchedulerClient = EventSchedulerClient(ns_client=self.notification_client)
 
     @expose('/health')
     def health(self):
@@ -1977,6 +1979,19 @@ def tree(self):
         node_count = 0
         node_limit = 5000 / max(1, len(dag.leaves))
 
+        upstream_tasks: Dict[str, Set] = {}
+        downstream_tasks: Set = set()
+        for t in dag.tasks:
+            if t.get_subscribed_events():
+                downstream_tasks.add(t)
+                for event_namespace, event_key, event_type, from_task_id in BaseSerialization._deserialize(t.get_subscribed_events()):
+                    downstream_task = dag.get_task(t.task_id)
+                    if from_task_id:
+                        if from_task_id in upstream_tasks:
+                            upstream_tasks[from_task_id].add(downstream_task)
+                        else:
+                            upstream_tasks[from_task_id] = {downstream_task}
+
         def encode_ti(task_instance: Optional[models.TaskInstance]) -> Optional[List]:
             if not task_instance:
                 return None
@@ -2005,21 +2020,24 @@ def recurse_nodes(task, visited):
             visited.add(task)
             task_id = task.task_id
 
+            task_downstream = upstream_tasks[task.task_id].union(
+                task.downstream_list) if task.task_id in upstream_tasks else task.downstream_list
+
             node = {
                 'name': task.task_id,
                 'instances': [encode_ti(task_instances.get((task_id, d))) for d in dates],
-                'num_dep': len(task.downstream_list),
+                'num_dep': len(task_downstream),
                 'operator': task.task_type,
                 'retries': task.retries,
                 'owner': task.owner,
                 'ui_color': task.ui_color,
             }
 
-            if task.downstream_list:
+            if task_downstream:
                 children = [
-                    recurse_nodes(t, visited)
-                    for t in task.downstream_list
-                    if node_count < node_limit or t not in visited
+                    recurse_nodes(d, visited)
+                    for d in task_downstream
+                    if node_count < node_limit or d not in visited
                 ]
 
                 # D3 tree uses children vs _children to define what is
@@ -2046,7 +2064,7 @@ def recurse_nodes(task, visited):
 
         data = {
             'name': '[DAG]',
-            'children': [recurse_nodes(t, set()) for t in dag.roots],
+            'children': [recurse_nodes(t, set()) for t in [n for n in dag.roots if n not in downstream_tasks]],
             'instances': [dag_runs.get(d) or {'execution_date': d.isoformat()} for d in dates],
         }
 
@@ -2110,8 +2128,7 @@ def graph(self, session=None):
 
         arrange = request.args.get('arrange', dag.orientation)
 
-        events = {}
-        nodes = task_group_to_dict(dag.task_group, events)
+        nodes = task_group_to_dict(dag.task_group)
         edges = dag_edges(dag)
 
         dt_nr_dr_data = get_date_time_num_runs_dag_runs_form_data(request, session, dag)
@@ -2135,14 +2152,21 @@ class GraphForm(DateTimeWithNumRunsWithDagRunsForm):
         form.execution_date.choices = dt_nr_dr_data['dr_choices']
 
         task_instances = {ti.task_id: alchemy_to_dict(ti) for ti in dag.get_task_instances(dttm, dttm)}
-        tasks = {
-            t.task_id: {
+        tasks: Dict[str, Dict[str, str]] = {}
+        events: Dict[str, Tuple[str, str, str]] = {}
+        for t in dag.tasks:
+            tasks[t.task_id] = {
                 'dag_id': t.dag_id,
                 'task_type': t.task_type,
                 'extra_links': t.extra_links,
             }
-            for t in dag.tasks
-        }
+            if t.get_subscribed_events():
+                for event_namespace, event_key, event_type, from_task_id in BaseSerialization._deserialize(
+                        t.get_subscribed_events()):
+                    event = '{},{},{}'.format(event_namespace, event_key, event_type)
+                    if event not in events:
+                        events[event] = (event_namespace, event_key, event_type)
+
         if not tasks:
             flash("No tasks found", "error")
         session.commit()
@@ -2169,9 +2193,12 @@ class GraphForm(DateTimeWithNumRunsWithDagRunsForm):
             root=root or '',
             task_instances=task_instances,
             tasks=tasks,
+            events={key: (value[0], value[1], value[2],
+                          len(self.notification_client.list_events(
+                              namespace=value[0], key=value[1], event_type=value[2])))
+                    for key, value in events.items()},
             nodes=nodes,
             edges=edges,
-            events=json.dumps(events),
             show_external_log_redirect=task_log_reader.supports_external_link,
             external_log_name=external_log_name,
             dag_run_state=dt_nr_dr_data['dr_state'],
@@ -2688,7 +2715,7 @@ def extra_links(self):
             response.status_code = 404
             return response
 
-    @expose('/object/task_instances')
+    @expose('/object/nodes')
     @auth.has_access(
         [
             (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG),
@@ -2698,7 +2725,7 @@ def extra_links(self):
         ]
     )
     @action_logging
-    def task_instances(self):
+    def nodes(self):
         """Shows task instances."""
         dag_id = request.args.get('dag_id')
         dag = current_app.dag_bag.get_dag(dag_id)
@@ -2711,7 +2738,20 @@ def task_instances(self):
 
         task_instances = {ti.task_id: alchemy_to_dict(ti) for ti in dag.get_task_instances(dttm, dttm)}
 
-        return json.dumps(task_instances, cls=utils_json.AirflowJsonEncoder)
+        events: Dict[str, Tuple[str, str, str]] = {}
+        for t in dag.tasks:
+            if t.get_subscribed_events():
+                for event_namespace, event_key, event_type, from_task_id in BaseSerialization._deserialize(
+                        t.get_subscribed_events()):
+                    event = '{},{},{}'.format(event_namespace, event_key, event_type)
+                    events[event] = (event_namespace, event_key, event_type)
+
+        return json.dumps({'task_instances': task_instances,
+                           'events': {key: (value[0], value[1], value[2],
+                                            len(self.notification_client.list_events(namespace=value[0], key=value[1],
+                                                                                     event_type=value[2])))
+                                      for key, value in events.items()}},
+                          cls=utils_json.AirflowJsonEncoder)
 
 
 class ConfigurationView(AirflowBaseView):
@@ -4022,11 +4062,11 @@ class EventModelView(AirflowModelView):
 
     page_size = PAGE_SIZE
 
-    list_columns = ['key', 'version', 'value', 'event_type', 'context', 'namespace', 'create_time', 'uuid']
+    list_columns = ['key', 'version', 'value', 'event_type', 'context', 'namespace', 'sender', 'create_time', 'uuid']
 
     order_columns = [item for item in list_columns if item not in ['context']]
 
-    search_columns = ['key', 'version', 'event_type', 'namespace', 'create_time']
+    search_columns = ['key', 'version', 'event_type', 'namespace', 'sender', 'create_time']
 
     base_order = ('key', 'asc')
 
diff --git a/flink-ai-flow/lib/airflow/airflow/www/webpack.config.js b/flink-ai-flow/lib/airflow/airflow/www/webpack.config.js
index f63aa0bfe..58ce91c9b 100644
--- a/flink-ai-flow/lib/airflow/airflow/www/webpack.config.js
+++ b/flink-ai-flow/lib/airflow/airflow/www/webpack.config.js
@@ -50,6 +50,7 @@ const config = {
     moment: 'moment-timezone',
     switch: `${STATIC_DIR}/css/switch.css`,
     taskInstances: `${STATIC_DIR}/js/task-instances.js`,
+    events: `${STATIC_DIR}/js/events.js`,
     tree: `${STATIC_DIR}/css/tree.css`,
   },
   output: {
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/proto/NotificationServiceOuterClass.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/proto/NotificationServiceOuterClass.java
index 5144a41a3..ad3c85832 100644
--- a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/proto/NotificationServiceOuterClass.java
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/proto/NotificationServiceOuterClass.java
@@ -102,10 +102,6 @@ public ReturnStatus findValueByNumber(int number) {
 
     public final com.google.protobuf.Descriptors.EnumValueDescriptor
         getValueDescriptor() {
-      if (this == UNRECOGNIZED) {
-        throw new java.lang.IllegalStateException(
-            "Can't get the descriptor of an unrecognized enum value.");
-      }
       return getDescriptor().getValues().get(ordinal());
     }
     public final com.google.protobuf.Descriptors.EnumDescriptor
@@ -215,11 +211,23 @@ public interface EventProtoOrBuilder extends
      * @return The createTime.
      */
     long getCreateTime();
+
+    /**
+     * <code>string sender = 8;</code>
+     * @return The sender.
+     */
+    java.lang.String getSender();
+    /**
+     * <code>string sender = 8;</code>
+     * @return The bytes for sender.
+     */
+    com.google.protobuf.ByteString
+        getSenderBytes();
   }
   /**
    * Protobuf type {@code notification_service.EventProto}
    */
-  public static final class EventProto extends
+  public  static final class EventProto extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.EventProto)
       EventProtoOrBuilder {
@@ -234,6 +242,7 @@ private EventProto() {
       eventType_ = "";
       context_ = "";
       namespace_ = "";
+      sender_ = "";
     }
 
     @java.lang.Override
@@ -306,6 +315,12 @@ private EventProto(
               createTime_ = input.readInt64();
               break;
             }
+            case 66: {
+              java.lang.String s = input.readStringRequireUtf8();
+
+              sender_ = s;
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -344,7 +359,6 @@ private EventProto(
      * <code>string key = 1;</code>
      * @return The key.
      */
-    @java.lang.Override
     public java.lang.String getKey() {
       java.lang.Object ref = key_;
       if (ref instanceof java.lang.String) {
@@ -361,7 +375,6 @@ public java.lang.String getKey() {
      * <code>string key = 1;</code>
      * @return The bytes for key.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getKeyBytes() {
       java.lang.Object ref = key_;
@@ -382,7 +395,6 @@ public java.lang.String getKey() {
      * <code>string value = 2;</code>
      * @return The value.
      */
-    @java.lang.Override
     public java.lang.String getValue() {
       java.lang.Object ref = value_;
       if (ref instanceof java.lang.String) {
@@ -399,7 +411,6 @@ public java.lang.String getValue() {
      * <code>string value = 2;</code>
      * @return The bytes for value.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getValueBytes() {
       java.lang.Object ref = value_;
@@ -420,7 +431,6 @@ public java.lang.String getValue() {
      * <code>string event_type = 3;</code>
      * @return The eventType.
      */
-    @java.lang.Override
     public java.lang.String getEventType() {
       java.lang.Object ref = eventType_;
       if (ref instanceof java.lang.String) {
@@ -437,7 +447,6 @@ public java.lang.String getEventType() {
      * <code>string event_type = 3;</code>
      * @return The bytes for eventType.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getEventTypeBytes() {
       java.lang.Object ref = eventType_;
@@ -458,7 +467,6 @@ public java.lang.String getEventType() {
      * <code>string context = 4;</code>
      * @return The context.
      */
-    @java.lang.Override
     public java.lang.String getContext() {
       java.lang.Object ref = context_;
       if (ref instanceof java.lang.String) {
@@ -475,7 +483,6 @@ public java.lang.String getContext() {
      * <code>string context = 4;</code>
      * @return The bytes for context.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getContextBytes() {
       java.lang.Object ref = context_;
@@ -496,7 +503,6 @@ public java.lang.String getContext() {
      * <code>string namespace = 5;</code>
      * @return The namespace.
      */
-    @java.lang.Override
     public java.lang.String getNamespace() {
       java.lang.Object ref = namespace_;
       if (ref instanceof java.lang.String) {
@@ -513,7 +519,6 @@ public java.lang.String getNamespace() {
      * <code>string namespace = 5;</code>
      * @return The bytes for namespace.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getNamespaceBytes() {
       java.lang.Object ref = namespace_;
@@ -534,7 +539,6 @@ public java.lang.String getNamespace() {
      * <code>int64 version = 6;</code>
      * @return The version.
      */
-    @java.lang.Override
     public long getVersion() {
       return version_;
     }
@@ -545,11 +549,46 @@ public long getVersion() {
      * <code>int64 create_time = 7;</code>
      * @return The createTime.
      */
-    @java.lang.Override
     public long getCreateTime() {
       return createTime_;
     }
 
+    public static final int SENDER_FIELD_NUMBER = 8;
+    private volatile java.lang.Object sender_;
+    /**
+     * <code>string sender = 8;</code>
+     * @return The sender.
+     */
+    public java.lang.String getSender() {
+      java.lang.Object ref = sender_;
+      if (ref instanceof java.lang.String) {
+        return (java.lang.String) ref;
+      } else {
+        com.google.protobuf.ByteString bs = 
+            (com.google.protobuf.ByteString) ref;
+        java.lang.String s = bs.toStringUtf8();
+        sender_ = s;
+        return s;
+      }
+    }
+    /**
+     * <code>string sender = 8;</code>
+     * @return The bytes for sender.
+     */
+    public com.google.protobuf.ByteString
+        getSenderBytes() {
+      java.lang.Object ref = sender_;
+      if (ref instanceof java.lang.String) {
+        com.google.protobuf.ByteString b = 
+            com.google.protobuf.ByteString.copyFromUtf8(
+                (java.lang.String) ref);
+        sender_ = b;
+        return b;
+      } else {
+        return (com.google.protobuf.ByteString) ref;
+      }
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -585,6 +624,9 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
       if (createTime_ != 0L) {
         output.writeInt64(7, createTime_);
       }
+      if (!getSenderBytes().isEmpty()) {
+        com.google.protobuf.GeneratedMessageV3.writeString(output, 8, sender_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -617,6 +659,9 @@ public int getSerializedSize() {
         size += com.google.protobuf.CodedOutputStream
           .computeInt64Size(7, createTime_);
       }
+      if (!getSenderBytes().isEmpty()) {
+        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(8, sender_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -646,6 +691,8 @@ public boolean equals(final java.lang.Object obj) {
           != other.getVersion()) return false;
       if (getCreateTime()
           != other.getCreateTime()) return false;
+      if (!getSender()
+          .equals(other.getSender())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -673,6 +720,8 @@ public int hashCode() {
       hash = (37 * hash) + CREATE_TIME_FIELD_NUMBER;
       hash = (53 * hash) + com.google.protobuf.Internal.hashLong(
           getCreateTime());
+      hash = (37 * hash) + SENDER_FIELD_NUMBER;
+      hash = (53 * hash) + getSender().hashCode();
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -820,6 +869,8 @@ public Builder clear() {
 
         createTime_ = 0L;
 
+        sender_ = "";
+
         return this;
       }
 
@@ -853,6 +904,7 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto bu
         result.namespace_ = namespace_;
         result.version_ = version_;
         result.createTime_ = createTime_;
+        result.sender_ = sender_;
         onBuilt();
         return result;
       }
@@ -927,6 +979,10 @@ public Builder mergeFrom(com.aiflow.notification.proto.NotificationServiceOuterC
         if (other.getCreateTime() != 0L) {
           setCreateTime(other.getCreateTime());
         }
+        if (!other.getSender().isEmpty()) {
+          sender_ = other.sender_;
+          onChanged();
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -1341,7 +1397,6 @@ public Builder setNamespaceBytes(
        * <code>int64 version = 6;</code>
        * @return The version.
        */
-      @java.lang.Override
       public long getVersion() {
         return version_;
       }
@@ -1372,7 +1427,6 @@ public Builder clearVersion() {
        * <code>int64 create_time = 7;</code>
        * @return The createTime.
        */
-      @java.lang.Override
       public long getCreateTime() {
         return createTime_;
       }
@@ -1397,6 +1451,82 @@ public Builder clearCreateTime() {
         onChanged();
         return this;
       }
+
+      private java.lang.Object sender_ = "";
+      /**
+       * <code>string sender = 8;</code>
+       * @return The sender.
+       */
+      public java.lang.String getSender() {
+        java.lang.Object ref = sender_;
+        if (!(ref instanceof java.lang.String)) {
+          com.google.protobuf.ByteString bs =
+              (com.google.protobuf.ByteString) ref;
+          java.lang.String s = bs.toStringUtf8();
+          sender_ = s;
+          return s;
+        } else {
+          return (java.lang.String) ref;
+        }
+      }
+      /**
+       * <code>string sender = 8;</code>
+       * @return The bytes for sender.
+       */
+      public com.google.protobuf.ByteString
+          getSenderBytes() {
+        java.lang.Object ref = sender_;
+        if (ref instanceof String) {
+          com.google.protobuf.ByteString b = 
+              com.google.protobuf.ByteString.copyFromUtf8(
+                  (java.lang.String) ref);
+          sender_ = b;
+          return b;
+        } else {
+          return (com.google.protobuf.ByteString) ref;
+        }
+      }
+      /**
+       * <code>string sender = 8;</code>
+       * @param value The sender to set.
+       * @return This builder for chaining.
+       */
+      public Builder setSender(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  
+        sender_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>string sender = 8;</code>
+       * @return This builder for chaining.
+       */
+      public Builder clearSender() {
+        
+        sender_ = getDefaultInstance().getSender();
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>string sender = 8;</code>
+       * @param value The bytes for sender to set.
+       * @return This builder for chaining.
+       */
+      public Builder setSenderBytes(
+          com.google.protobuf.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  checkByteStringIsUtf8(value);
+        
+        sender_ = value;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -1492,7 +1622,7 @@ public interface SendEventRequestOrBuilder extends
   /**
    * Protobuf type {@code notification_service.SendEventRequest}
    */
-  public static final class SendEventRequest extends
+  public  static final class SendEventRequest extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.SendEventRequest)
       SendEventRequestOrBuilder {
@@ -1592,7 +1722,6 @@ private SendEventRequest(
      * <code>.notification_service.EventProto event = 1;</code>
      * @return Whether the event field is set.
      */
-    @java.lang.Override
     public boolean hasEvent() {
       return event_ != null;
     }
@@ -1600,14 +1729,12 @@ public boolean hasEvent() {
      * <code>.notification_service.EventProto event = 1;</code>
      * @return The event.
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto getEvent() {
       return event_ == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto.getDefaultInstance() : event_;
     }
     /**
      * <code>.notification_service.EventProto event = 1;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrBuilder getEventOrBuilder() {
       return getEvent();
     }
@@ -1622,7 +1749,6 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrB
      * <code>string uuid = 2;</code>
      * @return The uuid.
      */
-    @java.lang.Override
     public java.lang.String getUuid() {
       java.lang.Object ref = uuid_;
       if (ref instanceof java.lang.String) {
@@ -1643,7 +1769,6 @@ public java.lang.String getUuid() {
      * <code>string uuid = 2;</code>
      * @return The bytes for uuid.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getUuidBytes() {
       java.lang.Object ref = uuid_;
@@ -2302,7 +2427,7 @@ public interface SendEventsResponseOrBuilder extends
   /**
    * Protobuf type {@code notification_service.SendEventsResponse}
    */
-  public static final class SendEventsResponse extends
+  public  static final class SendEventsResponse extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.SendEventsResponse)
       SendEventsResponseOrBuilder {
@@ -2409,7 +2534,6 @@ private SendEventsResponse(
      * <code>.notification_service.EventProto event = 1;</code>
      * @return Whether the event field is set.
      */
-    @java.lang.Override
     public boolean hasEvent() {
       return event_ != null;
     }
@@ -2417,14 +2541,12 @@ public boolean hasEvent() {
      * <code>.notification_service.EventProto event = 1;</code>
      * @return The event.
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto getEvent() {
       return event_ == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto.getDefaultInstance() : event_;
     }
     /**
      * <code>.notification_service.EventProto event = 1;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrBuilder getEventOrBuilder() {
       return getEvent();
     }
@@ -2435,14 +2557,14 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrB
      * <code>.notification_service.ReturnStatus return_code = 2;</code>
      * @return The enum numeric value on the wire for returnCode.
      */
-    @java.lang.Override public int getReturnCodeValue() {
+    public int getReturnCodeValue() {
       return returnCode_;
     }
     /**
      * <code>.notification_service.ReturnStatus return_code = 2;</code>
      * @return The returnCode.
      */
-    @java.lang.Override public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
+    public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
       @SuppressWarnings("deprecation")
       com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
       return result == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.UNRECOGNIZED : result;
@@ -2454,7 +2576,6 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrB
      * <code>string return_msg = 3;</code>
      * @return The returnMsg.
      */
-    @java.lang.Override
     public java.lang.String getReturnMsg() {
       java.lang.Object ref = returnMsg_;
       if (ref instanceof java.lang.String) {
@@ -2471,7 +2592,6 @@ public java.lang.String getReturnMsg() {
      * <code>string return_msg = 3;</code>
      * @return The bytes for returnMsg.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getReturnMsgBytes() {
       java.lang.Object ref = returnMsg_;
@@ -2958,7 +3078,7 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrB
        * <code>.notification_service.ReturnStatus return_code = 2;</code>
        * @return The enum numeric value on the wire for returnCode.
        */
-      @java.lang.Override public int getReturnCodeValue() {
+      public int getReturnCodeValue() {
         return returnCode_;
       }
       /**
@@ -2967,7 +3087,6 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrB
        * @return This builder for chaining.
        */
       public Builder setReturnCodeValue(int value) {
-        
         returnCode_ = value;
         onChanged();
         return this;
@@ -2976,7 +3095,6 @@ public Builder setReturnCodeValue(int value) {
        * <code>.notification_service.ReturnStatus return_code = 2;</code>
        * @return The returnCode.
        */
-      @java.lang.Override
       public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
         @SuppressWarnings("deprecation")
         com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
@@ -3205,11 +3323,23 @@ public interface ListEventsRequestOrBuilder extends
      */
     com.google.protobuf.ByteString
         getNamespaceBytes();
+
+    /**
+     * <code>string sender = 7;</code>
+     * @return The sender.
+     */
+    java.lang.String getSender();
+    /**
+     * <code>string sender = 7;</code>
+     * @return The bytes for sender.
+     */
+    com.google.protobuf.ByteString
+        getSenderBytes();
   }
   /**
    * Protobuf type {@code notification_service.ListEventsRequest}
    */
-  public static final class ListEventsRequest extends
+  public  static final class ListEventsRequest extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.ListEventsRequest)
       ListEventsRequestOrBuilder {
@@ -3222,6 +3352,7 @@ private ListEventsRequest() {
       keys_ = com.google.protobuf.LazyStringArrayList.EMPTY;
       eventType_ = "";
       namespace_ = "";
+      sender_ = "";
     }
 
     @java.lang.Override
@@ -3291,6 +3422,12 @@ private ListEventsRequest(
               namespace_ = s;
               break;
             }
+            case 58: {
+              java.lang.String s = input.readStringRequireUtf8();
+
+              sender_ = s;
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -3367,7 +3504,6 @@ public java.lang.String getKeys(int index) {
      * <code>string event_type = 2;</code>
      * @return The eventType.
      */
-    @java.lang.Override
     public java.lang.String getEventType() {
       java.lang.Object ref = eventType_;
       if (ref instanceof java.lang.String) {
@@ -3384,7 +3520,6 @@ public java.lang.String getEventType() {
      * <code>string event_type = 2;</code>
      * @return The bytes for eventType.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getEventTypeBytes() {
       java.lang.Object ref = eventType_;
@@ -3405,7 +3540,6 @@ public java.lang.String getEventType() {
      * <code>int64 start_time = 3;</code>
      * @return The startTime.
      */
-    @java.lang.Override
     public long getStartTime() {
       return startTime_;
     }
@@ -3416,7 +3550,6 @@ public long getStartTime() {
      * <code>int64 start_version = 4;</code>
      * @return The startVersion.
      */
-    @java.lang.Override
     public long getStartVersion() {
       return startVersion_;
     }
@@ -3427,7 +3560,6 @@ public long getStartVersion() {
      * <code>int32 timeout_seconds = 5;</code>
      * @return The timeoutSeconds.
      */
-    @java.lang.Override
     public int getTimeoutSeconds() {
       return timeoutSeconds_;
     }
@@ -3438,7 +3570,6 @@ public int getTimeoutSeconds() {
      * <code>string namespace = 6;</code>
      * @return The namespace.
      */
-    @java.lang.Override
     public java.lang.String getNamespace() {
       java.lang.Object ref = namespace_;
       if (ref instanceof java.lang.String) {
@@ -3455,7 +3586,6 @@ public java.lang.String getNamespace() {
      * <code>string namespace = 6;</code>
      * @return The bytes for namespace.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getNamespaceBytes() {
       java.lang.Object ref = namespace_;
@@ -3470,6 +3600,42 @@ public java.lang.String getNamespace() {
       }
     }
 
+    public static final int SENDER_FIELD_NUMBER = 7;
+    private volatile java.lang.Object sender_;
+    /**
+     * <code>string sender = 7;</code>
+     * @return The sender.
+     */
+    public java.lang.String getSender() {
+      java.lang.Object ref = sender_;
+      if (ref instanceof java.lang.String) {
+        return (java.lang.String) ref;
+      } else {
+        com.google.protobuf.ByteString bs = 
+            (com.google.protobuf.ByteString) ref;
+        java.lang.String s = bs.toStringUtf8();
+        sender_ = s;
+        return s;
+      }
+    }
+    /**
+     * <code>string sender = 7;</code>
+     * @return The bytes for sender.
+     */
+    public com.google.protobuf.ByteString
+        getSenderBytes() {
+      java.lang.Object ref = sender_;
+      if (ref instanceof java.lang.String) {
+        com.google.protobuf.ByteString b = 
+            com.google.protobuf.ByteString.copyFromUtf8(
+                (java.lang.String) ref);
+        sender_ = b;
+        return b;
+      } else {
+        return (com.google.protobuf.ByteString) ref;
+      }
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -3502,6 +3668,9 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
       if (!getNamespaceBytes().isEmpty()) {
         com.google.protobuf.GeneratedMessageV3.writeString(output, 6, namespace_);
       }
+      if (!getSenderBytes().isEmpty()) {
+        com.google.protobuf.GeneratedMessageV3.writeString(output, 7, sender_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -3537,6 +3706,9 @@ public int getSerializedSize() {
       if (!getNamespaceBytes().isEmpty()) {
         size += com.google.protobuf.GeneratedMessageV3.computeStringSize(6, namespace_);
       }
+      if (!getSenderBytes().isEmpty()) {
+        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(7, sender_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -3564,6 +3736,8 @@ public boolean equals(final java.lang.Object obj) {
           != other.getTimeoutSeconds()) return false;
       if (!getNamespace()
           .equals(other.getNamespace())) return false;
+      if (!getSender()
+          .equals(other.getSender())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -3591,6 +3765,8 @@ public int hashCode() {
       hash = (53 * hash) + getTimeoutSeconds();
       hash = (37 * hash) + NAMESPACE_FIELD_NUMBER;
       hash = (53 * hash) + getNamespace().hashCode();
+      hash = (37 * hash) + SENDER_FIELD_NUMBER;
+      hash = (53 * hash) + getSender().hashCode();
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -3736,6 +3912,8 @@ public Builder clear() {
 
         namespace_ = "";
 
+        sender_ = "";
+
         return this;
       }
 
@@ -3773,6 +3951,7 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.ListEventsReq
         result.startVersion_ = startVersion_;
         result.timeoutSeconds_ = timeoutSeconds_;
         result.namespace_ = namespace_;
+        result.sender_ = sender_;
         onBuilt();
         return result;
       }
@@ -3848,6 +4027,10 @@ public Builder mergeFrom(com.aiflow.notification.proto.NotificationServiceOuterC
           namespace_ = other.namespace_;
           onChanged();
         }
+        if (!other.getSender().isEmpty()) {
+          sender_ = other.sender_;
+          onChanged();
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -4069,7 +4252,6 @@ public Builder setEventTypeBytes(
        * <code>int64 start_time = 3;</code>
        * @return The startTime.
        */
-      @java.lang.Override
       public long getStartTime() {
         return startTime_;
       }
@@ -4100,7 +4282,6 @@ public Builder clearStartTime() {
        * <code>int64 start_version = 4;</code>
        * @return The startVersion.
        */
-      @java.lang.Override
       public long getStartVersion() {
         return startVersion_;
       }
@@ -4131,7 +4312,6 @@ public Builder clearStartVersion() {
        * <code>int32 timeout_seconds = 5;</code>
        * @return The timeoutSeconds.
        */
-      @java.lang.Override
       public int getTimeoutSeconds() {
         return timeoutSeconds_;
       }
@@ -4232,6 +4412,82 @@ public Builder setNamespaceBytes(
         onChanged();
         return this;
       }
+
+      private java.lang.Object sender_ = "";
+      /**
+       * <code>string sender = 7;</code>
+       * @return The sender.
+       */
+      public java.lang.String getSender() {
+        java.lang.Object ref = sender_;
+        if (!(ref instanceof java.lang.String)) {
+          com.google.protobuf.ByteString bs =
+              (com.google.protobuf.ByteString) ref;
+          java.lang.String s = bs.toStringUtf8();
+          sender_ = s;
+          return s;
+        } else {
+          return (java.lang.String) ref;
+        }
+      }
+      /**
+       * <code>string sender = 7;</code>
+       * @return The bytes for sender.
+       */
+      public com.google.protobuf.ByteString
+          getSenderBytes() {
+        java.lang.Object ref = sender_;
+        if (ref instanceof String) {
+          com.google.protobuf.ByteString b = 
+              com.google.protobuf.ByteString.copyFromUtf8(
+                  (java.lang.String) ref);
+          sender_ = b;
+          return b;
+        } else {
+          return (com.google.protobuf.ByteString) ref;
+        }
+      }
+      /**
+       * <code>string sender = 7;</code>
+       * @param value The sender to set.
+       * @return This builder for chaining.
+       */
+      public Builder setSender(
+          java.lang.String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  
+        sender_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>string sender = 7;</code>
+       * @return This builder for chaining.
+       */
+      public Builder clearSender() {
+        
+        sender_ = getDefaultInstance().getSender();
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>string sender = 7;</code>
+       * @param value The bytes for sender to set.
+       * @return This builder for chaining.
+       */
+      public Builder setSenderBytes(
+          com.google.protobuf.ByteString value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  checkByteStringIsUtf8(value);
+        
+        sender_ = value;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -4316,7 +4572,7 @@ public interface ListAllEventsRequestOrBuilder extends
   /**
    * Protobuf type {@code notification_service.ListAllEventsRequest}
    */
-  public static final class ListAllEventsRequest extends
+  public  static final class ListAllEventsRequest extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.ListAllEventsRequest)
       ListAllEventsRequestOrBuilder {
@@ -4416,7 +4672,6 @@ private ListAllEventsRequest(
      * <code>int32 timeout_seconds = 1;</code>
      * @return The timeoutSeconds.
      */
-    @java.lang.Override
     public int getTimeoutSeconds() {
       return timeoutSeconds_;
     }
@@ -4427,7 +4682,6 @@ public int getTimeoutSeconds() {
      * <code>int64 start_time = 2;</code>
      * @return The startTime.
      */
-    @java.lang.Override
     public long getStartTime() {
       return startTime_;
     }
@@ -4438,7 +4692,6 @@ public long getStartTime() {
      * <code>int64 start_version = 3;</code>
      * @return The startVersion.
      */
-    @java.lang.Override
     public long getStartVersion() {
       return startVersion_;
     }
@@ -4449,7 +4702,6 @@ public long getStartVersion() {
      * <code>int64 end_version = 4;</code>
      * @return The endVersion.
      */
-    @java.lang.Override
     public long getEndVersion() {
       return endVersion_;
     }
@@ -4815,7 +5067,6 @@ public Builder mergeFrom(
        * <code>int32 timeout_seconds = 1;</code>
        * @return The timeoutSeconds.
        */
-      @java.lang.Override
       public int getTimeoutSeconds() {
         return timeoutSeconds_;
       }
@@ -4846,7 +5097,6 @@ public Builder clearTimeoutSeconds() {
        * <code>int64 start_time = 2;</code>
        * @return The startTime.
        */
-      @java.lang.Override
       public long getStartTime() {
         return startTime_;
       }
@@ -4877,7 +5127,6 @@ public Builder clearStartTime() {
        * <code>int64 start_version = 3;</code>
        * @return The startVersion.
        */
-      @java.lang.Override
       public long getStartVersion() {
         return startVersion_;
       }
@@ -4908,7 +5157,6 @@ public Builder clearStartVersion() {
        * <code>int64 end_version = 4;</code>
        * @return The endVersion.
        */
-      @java.lang.Override
       public long getEndVersion() {
         return endVersion_;
       }
@@ -5040,7 +5288,7 @@ com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrBuilder
   /**
    * Protobuf type {@code notification_service.ListEventsResponse}
    */
-  public static final class ListEventsResponse extends
+  public  static final class ListEventsResponse extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.ListEventsResponse)
       ListEventsResponseOrBuilder {
@@ -5148,14 +5396,14 @@ private ListEventsResponse(
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The enum numeric value on the wire for returnCode.
      */
-    @java.lang.Override public int getReturnCodeValue() {
+    public int getReturnCodeValue() {
       return returnCode_;
     }
     /**
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The returnCode.
      */
-    @java.lang.Override public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
+    public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
       @SuppressWarnings("deprecation")
       com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
       return result == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.UNRECOGNIZED : result;
@@ -5167,7 +5415,6 @@ private ListEventsResponse(
      * <code>string return_msg = 2;</code>
      * @return The returnMsg.
      */
-    @java.lang.Override
     public java.lang.String getReturnMsg() {
       java.lang.Object ref = returnMsg_;
       if (ref instanceof java.lang.String) {
@@ -5184,7 +5431,6 @@ public java.lang.String getReturnMsg() {
      * <code>string return_msg = 2;</code>
      * @return The bytes for returnMsg.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getReturnMsgBytes() {
       java.lang.Object ref = returnMsg_;
@@ -5204,14 +5450,12 @@ public java.lang.String getReturnMsg() {
     /**
      * <code>repeated .notification_service.EventProto events = 3;</code>
      */
-    @java.lang.Override
     public java.util.List<com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto> getEventsList() {
       return events_;
     }
     /**
      * <code>repeated .notification_service.EventProto events = 3;</code>
      */
-    @java.lang.Override
     public java.util.List<? extends com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrBuilder> 
         getEventsOrBuilderList() {
       return events_;
@@ -5219,21 +5463,18 @@ public java.util.List<com.aiflow.notification.proto.NotificationServiceOuterClas
     /**
      * <code>repeated .notification_service.EventProto events = 3;</code>
      */
-    @java.lang.Override
     public int getEventsCount() {
       return events_.size();
     }
     /**
      * <code>repeated .notification_service.EventProto events = 3;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto getEvents(int index) {
       return events_.get(index);
     }
     /**
      * <code>repeated .notification_service.EventProto events = 3;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.EventProtoOrBuilder getEventsOrBuilder(
         int index) {
       return events_.get(index);
@@ -5619,7 +5860,7 @@ public Builder mergeFrom(
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The enum numeric value on the wire for returnCode.
        */
-      @java.lang.Override public int getReturnCodeValue() {
+      public int getReturnCodeValue() {
         return returnCode_;
       }
       /**
@@ -5628,7 +5869,6 @@ public Builder mergeFrom(
        * @return This builder for chaining.
        */
       public Builder setReturnCodeValue(int value) {
-        
         returnCode_ = value;
         onChanged();
         return this;
@@ -5637,7 +5877,6 @@ public Builder setReturnCodeValue(int value) {
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The returnCode.
        */
-      @java.lang.Override
       public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
         @SuppressWarnings("deprecation")
         com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
@@ -6073,7 +6312,7 @@ public interface GetLatestVersionByKeyRequestOrBuilder extends
   /**
    * Protobuf type {@code notification_service.GetLatestVersionByKeyRequest}
    */
-  public static final class GetLatestVersionByKeyRequest extends
+  public  static final class GetLatestVersionByKeyRequest extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.GetLatestVersionByKeyRequest)
       GetLatestVersionByKeyRequestOrBuilder {
@@ -6172,7 +6411,6 @@ private GetLatestVersionByKeyRequest(
      * <code>string key = 1;</code>
      * @return The key.
      */
-    @java.lang.Override
     public java.lang.String getKey() {
       java.lang.Object ref = key_;
       if (ref instanceof java.lang.String) {
@@ -6189,7 +6427,6 @@ public java.lang.String getKey() {
      * <code>string key = 1;</code>
      * @return The bytes for key.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getKeyBytes() {
       java.lang.Object ref = key_;
@@ -6210,7 +6447,6 @@ public java.lang.String getKey() {
      * <code>string namespace = 2;</code>
      * @return The namespace.
      */
-    @java.lang.Override
     public java.lang.String getNamespace() {
       java.lang.Object ref = namespace_;
       if (ref instanceof java.lang.String) {
@@ -6227,7 +6463,6 @@ public java.lang.String getNamespace() {
      * <code>string namespace = 2;</code>
      * @return The bytes for namespace.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getNamespaceBytes() {
       java.lang.Object ref = namespace_;
@@ -6248,7 +6483,6 @@ public java.lang.String getNamespace() {
      * <code>int32 timeout_seconds = 3;</code>
      * @return The timeoutSeconds.
      */
-    @java.lang.Override
     public int getTimeoutSeconds() {
       return timeoutSeconds_;
     }
@@ -6746,7 +6980,6 @@ public Builder setNamespaceBytes(
        * <code>int32 timeout_seconds = 3;</code>
        * @return The timeoutSeconds.
        */
-      @java.lang.Override
       public int getTimeoutSeconds() {
         return timeoutSeconds_;
       }
@@ -6861,7 +7094,7 @@ public interface GetLatestVersionResponseOrBuilder extends
   /**
    * Protobuf type {@code notification_service.GetLatestVersionResponse}
    */
-  public static final class GetLatestVersionResponse extends
+  public  static final class GetLatestVersionResponse extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.GetLatestVersionResponse)
       GetLatestVersionResponseOrBuilder {
@@ -6960,7 +7193,6 @@ private GetLatestVersionResponse(
      * <code>string return_code = 1;</code>
      * @return The returnCode.
      */
-    @java.lang.Override
     public java.lang.String getReturnCode() {
       java.lang.Object ref = returnCode_;
       if (ref instanceof java.lang.String) {
@@ -6977,7 +7209,6 @@ public java.lang.String getReturnCode() {
      * <code>string return_code = 1;</code>
      * @return The bytes for returnCode.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getReturnCodeBytes() {
       java.lang.Object ref = returnCode_;
@@ -6998,7 +7229,6 @@ public java.lang.String getReturnCode() {
      * <code>string return_msg = 2;</code>
      * @return The returnMsg.
      */
-    @java.lang.Override
     public java.lang.String getReturnMsg() {
       java.lang.Object ref = returnMsg_;
       if (ref instanceof java.lang.String) {
@@ -7015,7 +7245,6 @@ public java.lang.String getReturnMsg() {
      * <code>string return_msg = 2;</code>
      * @return The bytes for returnMsg.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getReturnMsgBytes() {
       java.lang.Object ref = returnMsg_;
@@ -7036,7 +7265,6 @@ public java.lang.String getReturnMsg() {
      * <code>int64 version = 3;</code>
      * @return The version.
      */
-    @java.lang.Override
     public long getVersion() {
       return version_;
     }
@@ -7535,7 +7763,6 @@ public Builder setReturnMsgBytes(
        * <code>int64 version = 3;</code>
        * @return The version.
        */
-      @java.lang.Override
       public long getVersion() {
         return version_;
       }
@@ -7644,7 +7871,7 @@ public interface MemberProtoOrBuilder extends
   /**
    * Protobuf type {@code notification_service.MemberProto}
    */
-  public static final class MemberProto extends
+  public  static final class MemberProto extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.MemberProto)
       MemberProtoOrBuilder {
@@ -7741,7 +7968,6 @@ private MemberProto(
      * <code>int64 version = 1;</code>
      * @return The version.
      */
-    @java.lang.Override
     public long getVersion() {
       return version_;
     }
@@ -7752,7 +7978,6 @@ public long getVersion() {
      * <code>string server_uri = 2;</code>
      * @return The serverUri.
      */
-    @java.lang.Override
     public java.lang.String getServerUri() {
       java.lang.Object ref = serverUri_;
       if (ref instanceof java.lang.String) {
@@ -7769,7 +7994,6 @@ public java.lang.String getServerUri() {
      * <code>string server_uri = 2;</code>
      * @return The bytes for serverUri.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getServerUriBytes() {
       java.lang.Object ref = serverUri_;
@@ -7790,7 +8014,6 @@ public java.lang.String getServerUri() {
      * <code>int64 update_time = 3;</code>
      * @return The updateTime.
      */
-    @java.lang.Override
     public long getUpdateTime() {
       return updateTime_;
     }
@@ -8138,7 +8361,6 @@ public Builder mergeFrom(
        * <code>int64 version = 1;</code>
        * @return The version.
        */
-      @java.lang.Override
       public long getVersion() {
         return version_;
       }
@@ -8245,7 +8467,6 @@ public Builder setServerUriBytes(
        * <code>int64 update_time = 3;</code>
        * @return The updateTime.
        */
-      @java.lang.Override
       public long getUpdateTime() {
         return updateTime_;
       }
@@ -8354,7 +8575,7 @@ public interface NotifyOrBuilder extends
   /**
    * Protobuf type {@code notification_service.Notify}
    */
-  public static final class Notify extends
+  public  static final class Notify extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.Notify)
       NotifyOrBuilder {
@@ -8448,7 +8669,6 @@ private Notify(
      * <code>string key = 1;</code>
      * @return The key.
      */
-    @java.lang.Override
     public java.lang.String getKey() {
       java.lang.Object ref = key_;
       if (ref instanceof java.lang.String) {
@@ -8465,7 +8685,6 @@ public java.lang.String getKey() {
      * <code>string key = 1;</code>
      * @return The bytes for key.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getKeyBytes() {
       java.lang.Object ref = key_;
@@ -8486,7 +8705,6 @@ public java.lang.String getKey() {
      * <code>string namespace = 2;</code>
      * @return The namespace.
      */
-    @java.lang.Override
     public java.lang.String getNamespace() {
       java.lang.Object ref = namespace_;
       if (ref instanceof java.lang.String) {
@@ -8503,7 +8721,6 @@ public java.lang.String getNamespace() {
      * <code>string namespace = 2;</code>
      * @return The bytes for namespace.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getNamespaceBytes() {
       java.lang.Object ref = namespace_;
@@ -9072,7 +9289,7 @@ com.aiflow.notification.proto.NotificationServiceOuterClass.NotifyOrBuilder getN
   /**
    * Protobuf type {@code notification_service.NotifyRequest}
    */
-  public static final class NotifyRequest extends
+  public  static final class NotifyRequest extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.NotifyRequest)
       NotifyRequestOrBuilder {
@@ -9165,14 +9382,12 @@ private NotifyRequest(
     /**
      * <code>repeated .notification_service.Notify notifies = 1;</code>
      */
-    @java.lang.Override
     public java.util.List<com.aiflow.notification.proto.NotificationServiceOuterClass.Notify> getNotifiesList() {
       return notifies_;
     }
     /**
      * <code>repeated .notification_service.Notify notifies = 1;</code>
      */
-    @java.lang.Override
     public java.util.List<? extends com.aiflow.notification.proto.NotificationServiceOuterClass.NotifyOrBuilder> 
         getNotifiesOrBuilderList() {
       return notifies_;
@@ -9180,21 +9395,18 @@ public java.util.List<com.aiflow.notification.proto.NotificationServiceOuterClas
     /**
      * <code>repeated .notification_service.Notify notifies = 1;</code>
      */
-    @java.lang.Override
     public int getNotifiesCount() {
       return notifies_.size();
     }
     /**
      * <code>repeated .notification_service.Notify notifies = 1;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.Notify getNotifies(int index) {
       return notifies_.get(index);
     }
     /**
      * <code>repeated .notification_service.Notify notifies = 1;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.NotifyOrBuilder getNotifiesOrBuilder(
         int index) {
       return notifies_.get(index);
@@ -9864,7 +10076,7 @@ public interface NotifyResponseOrBuilder extends
   /**
    * Protobuf type {@code notification_service.NotifyResponse}
    */
-  public static final class NotifyResponse extends
+  public  static final class NotifyResponse extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.NotifyResponse)
       NotifyResponseOrBuilder {
@@ -9958,14 +10170,14 @@ private NotifyResponse(
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The enum numeric value on the wire for returnCode.
      */
-    @java.lang.Override public int getReturnCodeValue() {
+    public int getReturnCodeValue() {
       return returnCode_;
     }
     /**
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The returnCode.
      */
-    @java.lang.Override public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
+    public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
       @SuppressWarnings("deprecation")
       com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
       return result == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.UNRECOGNIZED : result;
@@ -9977,7 +10189,6 @@ private NotifyResponse(
      * <code>string return_msg = 2;</code>
      * @return The returnMsg.
      */
-    @java.lang.Override
     public java.lang.String getReturnMsg() {
       java.lang.Object ref = returnMsg_;
       if (ref instanceof java.lang.String) {
@@ -9994,7 +10205,6 @@ public java.lang.String getReturnMsg() {
      * <code>string return_msg = 2;</code>
      * @return The bytes for returnMsg.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getReturnMsgBytes() {
       java.lang.Object ref = returnMsg_;
@@ -10332,7 +10542,7 @@ public Builder mergeFrom(
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The enum numeric value on the wire for returnCode.
        */
-      @java.lang.Override public int getReturnCodeValue() {
+      public int getReturnCodeValue() {
         return returnCode_;
       }
       /**
@@ -10341,7 +10551,6 @@ public Builder mergeFrom(
        * @return This builder for chaining.
        */
       public Builder setReturnCodeValue(int value) {
-        
         returnCode_ = value;
         onChanged();
         return this;
@@ -10350,7 +10559,6 @@ public Builder setReturnCodeValue(int value) {
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The returnCode.
        */
-      @java.lang.Override
       public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
         @SuppressWarnings("deprecation")
         com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
@@ -10522,7 +10730,7 @@ public interface ListMembersRequestOrBuilder extends
   /**
    * Protobuf type {@code notification_service.ListMembersRequest}
    */
-  public static final class ListMembersRequest extends
+  public  static final class ListMembersRequest extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.ListMembersRequest)
       ListMembersRequestOrBuilder {
@@ -10607,7 +10815,6 @@ private ListMembersRequest(
      * <code>int32 timeout_seconds = 1;</code>
      * @return The timeoutSeconds.
      */
-    @java.lang.Override
     public int getTimeoutSeconds() {
       return timeoutSeconds_;
     }
@@ -10919,7 +11126,6 @@ public Builder mergeFrom(
        * <code>int32 timeout_seconds = 1;</code>
        * @return The timeoutSeconds.
        */
-      @java.lang.Override
       public int getTimeoutSeconds() {
         return timeoutSeconds_;
       }
@@ -11051,7 +11257,7 @@ com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProtoOrBuilder
   /**
    * Protobuf type {@code notification_service.ListMembersResponse}
    */
-  public static final class ListMembersResponse extends
+  public  static final class ListMembersResponse extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.ListMembersResponse)
       ListMembersResponseOrBuilder {
@@ -11159,14 +11365,14 @@ private ListMembersResponse(
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The enum numeric value on the wire for returnCode.
      */
-    @java.lang.Override public int getReturnCodeValue() {
+    public int getReturnCodeValue() {
       return returnCode_;
     }
     /**
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The returnCode.
      */
-    @java.lang.Override public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
+    public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
       @SuppressWarnings("deprecation")
       com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
       return result == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.UNRECOGNIZED : result;
@@ -11178,7 +11384,6 @@ private ListMembersResponse(
      * <code>string return_msg = 2;</code>
      * @return The returnMsg.
      */
-    @java.lang.Override
     public java.lang.String getReturnMsg() {
       java.lang.Object ref = returnMsg_;
       if (ref instanceof java.lang.String) {
@@ -11195,7 +11400,6 @@ public java.lang.String getReturnMsg() {
      * <code>string return_msg = 2;</code>
      * @return The bytes for returnMsg.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getReturnMsgBytes() {
       java.lang.Object ref = returnMsg_;
@@ -11215,14 +11419,12 @@ public java.lang.String getReturnMsg() {
     /**
      * <code>repeated .notification_service.MemberProto members = 3;</code>
      */
-    @java.lang.Override
     public java.util.List<com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProto> getMembersList() {
       return members_;
     }
     /**
      * <code>repeated .notification_service.MemberProto members = 3;</code>
      */
-    @java.lang.Override
     public java.util.List<? extends com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProtoOrBuilder> 
         getMembersOrBuilderList() {
       return members_;
@@ -11230,21 +11432,18 @@ public java.util.List<com.aiflow.notification.proto.NotificationServiceOuterClas
     /**
      * <code>repeated .notification_service.MemberProto members = 3;</code>
      */
-    @java.lang.Override
     public int getMembersCount() {
       return members_.size();
     }
     /**
      * <code>repeated .notification_service.MemberProto members = 3;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProto getMembers(int index) {
       return members_.get(index);
     }
     /**
      * <code>repeated .notification_service.MemberProto members = 3;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProtoOrBuilder getMembersOrBuilder(
         int index) {
       return members_.get(index);
@@ -11630,7 +11829,7 @@ public Builder mergeFrom(
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The enum numeric value on the wire for returnCode.
        */
-      @java.lang.Override public int getReturnCodeValue() {
+      public int getReturnCodeValue() {
         return returnCode_;
       }
       /**
@@ -11639,7 +11838,6 @@ public Builder mergeFrom(
        * @return This builder for chaining.
        */
       public Builder setReturnCodeValue(int value) {
-        
         returnCode_ = value;
         onChanged();
         return this;
@@ -11648,7 +11846,6 @@ public Builder setReturnCodeValue(int value) {
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The returnCode.
        */
-      @java.lang.Override
       public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
         @SuppressWarnings("deprecation")
         com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
@@ -12069,7 +12266,7 @@ public interface NotifyNewMemberRequestOrBuilder extends
   /**
    * Protobuf type {@code notification_service.NotifyNewMemberRequest}
    */
-  public static final class NotifyNewMemberRequest extends
+  public  static final class NotifyNewMemberRequest extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.NotifyNewMemberRequest)
       NotifyNewMemberRequestOrBuilder {
@@ -12162,7 +12359,6 @@ private NotifyNewMemberRequest(
      * <code>.notification_service.MemberProto member = 1;</code>
      * @return Whether the member field is set.
      */
-    @java.lang.Override
     public boolean hasMember() {
       return member_ != null;
     }
@@ -12170,14 +12366,12 @@ public boolean hasMember() {
      * <code>.notification_service.MemberProto member = 1;</code>
      * @return The member.
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProto getMember() {
       return member_ == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProto.getDefaultInstance() : member_;
     }
     /**
      * <code>.notification_service.MemberProto member = 1;</code>
      */
-    @java.lang.Override
     public com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProtoOrBuilder getMemberOrBuilder() {
       return getMember();
     }
@@ -12698,7 +12892,7 @@ public interface NotifyNewMemberResponseOrBuilder extends
   /**
    * Protobuf type {@code notification_service.NotifyNewMemberResponse}
    */
-  public static final class NotifyNewMemberResponse extends
+  public  static final class NotifyNewMemberResponse extends
       com.google.protobuf.GeneratedMessageV3 implements
       // @@protoc_insertion_point(message_implements:notification_service.NotifyNewMemberResponse)
       NotifyNewMemberResponseOrBuilder {
@@ -12792,14 +12986,14 @@ private NotifyNewMemberResponse(
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The enum numeric value on the wire for returnCode.
      */
-    @java.lang.Override public int getReturnCodeValue() {
+    public int getReturnCodeValue() {
       return returnCode_;
     }
     /**
      * <code>.notification_service.ReturnStatus return_code = 1;</code>
      * @return The returnCode.
      */
-    @java.lang.Override public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
+    public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
       @SuppressWarnings("deprecation")
       com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
       return result == null ? com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.UNRECOGNIZED : result;
@@ -12811,7 +13005,6 @@ private NotifyNewMemberResponse(
      * <code>string return_msg = 2;</code>
      * @return The returnMsg.
      */
-    @java.lang.Override
     public java.lang.String getReturnMsg() {
       java.lang.Object ref = returnMsg_;
       if (ref instanceof java.lang.String) {
@@ -12828,7 +13021,6 @@ public java.lang.String getReturnMsg() {
      * <code>string return_msg = 2;</code>
      * @return The bytes for returnMsg.
      */
-    @java.lang.Override
     public com.google.protobuf.ByteString
         getReturnMsgBytes() {
       java.lang.Object ref = returnMsg_;
@@ -13166,7 +13358,7 @@ public Builder mergeFrom(
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The enum numeric value on the wire for returnCode.
        */
-      @java.lang.Override public int getReturnCodeValue() {
+      public int getReturnCodeValue() {
         return returnCode_;
       }
       /**
@@ -13175,7 +13367,6 @@ public Builder mergeFrom(
        * @return This builder for chaining.
        */
       public Builder setReturnCodeValue(int value) {
-        
         returnCode_ = value;
         onChanged();
         return this;
@@ -13184,7 +13375,6 @@ public Builder setReturnCodeValue(int value) {
        * <code>.notification_service.ReturnStatus return_code = 1;</code>
        * @return The returnCode.
        */
-      @java.lang.Override
       public com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus getReturnCode() {
         @SuppressWarnings("deprecation")
         com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus result = com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus.valueOf(returnCode_);
@@ -14155,77 +14345,79 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.GetLatestVers
     java.lang.String[] descriptorData = {
       "\n\032notification_service.proto\022\024notificati" +
       "on_service\032\034google/api/annotations.proto" +
-      "\"\206\001\n\nEventProto\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 " +
+      "\"\226\001\n\nEventProto\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002 " +
       "\001(\t\022\022\n\nevent_type\030\003 \001(\t\022\017\n\007context\030\004 \001(\t" +
       "\022\021\n\tnamespace\030\005 \001(\t\022\017\n\007version\030\006 \001(\003\022\023\n\013" +
-      "create_time\030\007 \001(\003\"Q\n\020SendEventRequest\022/\n" +
-      "\005event\030\001 \001(\0132 .notification_service.Even" +
-      "tProto\022\014\n\004uuid\030\002 \001(\t\"\222\001\n\022SendEventsRespo" +
-      "nse\022/\n\005event\030\001 \001(\0132 .notification_servic" +
-      "e.EventProto\0227\n\013return_code\030\002 \001(\0162\".noti" +
-      "fication_service.ReturnStatus\022\022\n\nreturn_" +
-      "msg\030\003 \001(\t\"\214\001\n\021ListEventsRequest\022\014\n\004keys\030" +
-      "\001 \003(\t\022\022\n\nevent_type\030\002 \001(\t\022\022\n\nstart_time\030" +
-      "\003 \001(\003\022\025\n\rstart_version\030\004 \001(\003\022\027\n\017timeout_" +
-      "seconds\030\005 \001(\005\022\021\n\tnamespace\030\006 \001(\t\"o\n\024List" +
-      "AllEventsRequest\022\027\n\017timeout_seconds\030\001 \001(" +
-      "\005\022\022\n\nstart_time\030\002 \001(\003\022\025\n\rstart_version\030\003" +
-      " \001(\003\022\023\n\013end_version\030\004 \001(\003\"\223\001\n\022ListEvents" +
+      "create_time\030\007 \001(\003\022\016\n\006sender\030\010 \001(\t\"Q\n\020Sen" +
+      "dEventRequest\022/\n\005event\030\001 \001(\0132 .notificat" +
+      "ion_service.EventProto\022\014\n\004uuid\030\002 \001(\t\"\222\001\n" +
+      "\022SendEventsResponse\022/\n\005event\030\001 \001(\0132 .not" +
+      "ification_service.EventProto\0227\n\013return_c" +
+      "ode\030\002 \001(\0162\".notification_service.ReturnS" +
+      "tatus\022\022\n\nreturn_msg\030\003 \001(\t\"\234\001\n\021ListEvents" +
+      "Request\022\014\n\004keys\030\001 \003(\t\022\022\n\nevent_type\030\002 \001(" +
+      "\t\022\022\n\nstart_time\030\003 \001(\003\022\025\n\rstart_version\030\004" +
+      " \001(\003\022\027\n\017timeout_seconds\030\005 \001(\005\022\021\n\tnamespa" +
+      "ce\030\006 \001(\t\022\016\n\006sender\030\007 \001(\t\"o\n\024ListAllEvent" +
+      "sRequest\022\027\n\017timeout_seconds\030\001 \001(\005\022\022\n\nsta" +
+      "rt_time\030\002 \001(\003\022\025\n\rstart_version\030\003 \001(\003\022\023\n\013" +
+      "end_version\030\004 \001(\003\"\223\001\n\022ListEventsResponse" +
+      "\0227\n\013return_code\030\001 \001(\0162\".notification_ser" +
+      "vice.ReturnStatus\022\022\n\nreturn_msg\030\002 \001(\t\0220\n" +
+      "\006events\030\003 \003(\0132 .notification_service.Eve" +
+      "ntProto\"W\n\034GetLatestVersionByKeyRequest\022" +
+      "\013\n\003key\030\001 \001(\t\022\021\n\tnamespace\030\002 \001(\t\022\027\n\017timeo" +
+      "ut_seconds\030\003 \001(\005\"T\n\030GetLatestVersionResp" +
+      "onse\022\023\n\013return_code\030\001 \001(\t\022\022\n\nreturn_msg\030" +
+      "\002 \001(\t\022\017\n\007version\030\003 \001(\003\"G\n\013MemberProto\022\017\n" +
+      "\007version\030\001 \001(\003\022\022\n\nserver_uri\030\002 \001(\t\022\023\n\013up" +
+      "date_time\030\003 \001(\003\"(\n\006Notify\022\013\n\003key\030\001 \001(\t\022\021" +
+      "\n\tnamespace\030\002 \001(\t\"?\n\rNotifyRequest\022.\n\010no" +
+      "tifies\030\001 \003(\0132\034.notification_service.Noti" +
+      "fy\"]\n\016NotifyResponse\0227\n\013return_code\030\001 \001(" +
+      "\0162\".notification_service.ReturnStatus\022\022\n" +
+      "\nreturn_msg\030\002 \001(\t\"-\n\022ListMembersRequest\022" +
+      "\027\n\017timeout_seconds\030\001 \001(\005\"\226\001\n\023ListMembers" +
       "Response\0227\n\013return_code\030\001 \001(\0162\".notifica" +
       "tion_service.ReturnStatus\022\022\n\nreturn_msg\030" +
-      "\002 \001(\t\0220\n\006events\030\003 \003(\0132 .notification_ser" +
-      "vice.EventProto\"W\n\034GetLatestVersionByKey" +
-      "Request\022\013\n\003key\030\001 \001(\t\022\021\n\tnamespace\030\002 \001(\t\022" +
-      "\027\n\017timeout_seconds\030\003 \001(\005\"T\n\030GetLatestVer" +
-      "sionResponse\022\023\n\013return_code\030\001 \001(\t\022\022\n\nret" +
-      "urn_msg\030\002 \001(\t\022\017\n\007version\030\003 \001(\003\"G\n\013Member" +
-      "Proto\022\017\n\007version\030\001 \001(\003\022\022\n\nserver_uri\030\002 \001" +
-      "(\t\022\023\n\013update_time\030\003 \001(\003\"(\n\006Notify\022\013\n\003key" +
-      "\030\001 \001(\t\022\021\n\tnamespace\030\002 \001(\t\"?\n\rNotifyReque" +
-      "st\022.\n\010notifies\030\001 \003(\0132\034.notification_serv" +
-      "ice.Notify\"]\n\016NotifyResponse\0227\n\013return_c" +
-      "ode\030\001 \001(\0162\".notification_service.ReturnS" +
-      "tatus\022\022\n\nreturn_msg\030\002 \001(\t\"-\n\022ListMembers" +
-      "Request\022\027\n\017timeout_seconds\030\001 \001(\005\"\226\001\n\023Lis" +
-      "tMembersResponse\0227\n\013return_code\030\001 \001(\0162\"." +
-      "notification_service.ReturnStatus\022\022\n\nret" +
-      "urn_msg\030\002 \001(\t\0222\n\007members\030\003 \003(\0132!.notific" +
-      "ation_service.MemberProto\"K\n\026NotifyNewMe" +
-      "mberRequest\0221\n\006member\030\001 \001(\0132!.notificati" +
-      "on_service.MemberProto\"f\n\027NotifyNewMembe" +
-      "rResponse\0227\n\013return_code\030\001 \001(\0162\".notific" +
-      "ation_service.ReturnStatus\022\022\n\nreturn_msg" +
-      "\030\002 \001(\t*&\n\014ReturnStatus\022\013\n\007SUCCESS\020\000\022\t\n\005E" +
-      "RROR\020\0012\357\010\n\023NotificationService\022\221\001\n\tsendE" +
-      "vent\022&.notification_service.SendEventReq" +
-      "uest\032(.notification_service.SendEventsRe" +
-      "sponse\"2\202\323\344\223\002,\"\'/aiflow/notification-ser" +
-      "vice/send_event:\001*\022\224\001\n\nlistEvents\022\'.noti" +
-      "fication_service.ListEventsRequest\032(.not" +
-      "ification_service.ListEventsResponse\"3\202\323" +
-      "\344\223\002-\"(/aiflow/notification-service/list_" +
-      "events:\001*\022\236\001\n\rlistAllEvents\022*.notificati" +
-      "on_service.ListAllEventsRequest\032(.notifi" +
-      "cation_service.ListEventsResponse\"7\202\323\344\223\002" +
-      "1\",/aiflow/notification-service/list_all" +
-      "_events:\001*\022\203\001\n\006notify\022#.notification_ser" +
-      "vice.NotifyRequest\032$.notification_servic" +
-      "e.NotifyResponse\".\202\323\344\223\002(\"#/aiflow/notifi" +
-      "cation-service/notify:\001*\022\230\001\n\013listMembers" +
-      "\022(.notification_service.ListMembersReque" +
-      "st\032).notification_service.ListMembersRes" +
-      "ponse\"4\202\323\344\223\002.\")/aiflow/notification-serv" +
-      "ice/list_members:\001*\022\251\001\n\017notifyNewMember\022" +
-      ",.notification_service.NotifyNewMemberRe" +
-      "quest\032-.notification_service.NotifyNewMe" +
-      "mberResponse\"9\202\323\344\223\0023\"./aiflow/notificati" +
-      "on-service/notify_new_member:\001*\022\276\001\n\025getL" +
-      "atestVersionByKey\0222.notification_service" +
-      ".GetLatestVersionByKeyRequest\032..notifica" +
-      "tion_service.GetLatestVersionResponse\"A\202" +
-      "\323\344\223\002;\"6/aiflow/notification-service/get_" +
-      "latest_version_by_key:\001*B%\n\035com.aiflow.n" +
-      "otification.proto\210\001\001\220\001\001b\006proto3"
+      "\002 \001(\t\0222\n\007members\030\003 \003(\0132!.notification_se" +
+      "rvice.MemberProto\"K\n\026NotifyNewMemberRequ" +
+      "est\0221\n\006member\030\001 \001(\0132!.notification_servi" +
+      "ce.MemberProto\"f\n\027NotifyNewMemberRespons" +
+      "e\0227\n\013return_code\030\001 \001(\0162\".notification_se" +
+      "rvice.ReturnStatus\022\022\n\nreturn_msg\030\002 \001(\t*&" +
+      "\n\014ReturnStatus\022\013\n\007SUCCESS\020\000\022\t\n\005ERROR\020\0012\357" +
+      "\010\n\023NotificationService\022\221\001\n\tsendEvent\022&.n" +
+      "otification_service.SendEventRequest\032(.n" +
+      "otification_service.SendEventsResponse\"2" +
+      "\202\323\344\223\002,\"\'/aiflow/notification-service/sen" +
+      "d_event:\001*\022\224\001\n\nlistEvents\022\'.notification" +
+      "_service.ListEventsRequest\032(.notificatio" +
+      "n_service.ListEventsResponse\"3\202\323\344\223\002-\"(/a" +
+      "iflow/notification-service/list_events:\001" +
+      "*\022\236\001\n\rlistAllEvents\022*.notification_servi" +
+      "ce.ListAllEventsRequest\032(.notification_s" +
+      "ervice.ListEventsResponse\"7\202\323\344\223\0021\",/aifl" +
+      "ow/notification-service/list_all_events:" +
+      "\001*\022\203\001\n\006notify\022#.notification_service.Not" +
+      "ifyRequest\032$.notification_service.Notify" +
+      "Response\".\202\323\344\223\002(\"#/aiflow/notification-s" +
+      "ervice/notify:\001*\022\230\001\n\013listMembers\022(.notif" +
+      "ication_service.ListMembersRequest\032).not" +
+      "ification_service.ListMembersResponse\"4\202" +
+      "\323\344\223\002.\")/aiflow/notification-service/list" +
+      "_members:\001*\022\251\001\n\017notifyNewMember\022,.notifi" +
+      "cation_service.NotifyNewMemberRequest\032-." +
+      "notification_service.NotifyNewMemberResp" +
+      "onse\"9\202\323\344\223\0023\"./aiflow/notification-servi" +
+      "ce/notify_new_member:\001*\022\276\001\n\025getLatestVer" +
+      "sionByKey\0222.notification_service.GetLate" +
+      "stVersionByKeyRequest\032..notification_ser" +
+      "vice.GetLatestVersionResponse\"A\202\323\344\223\002;\"6/" +
+      "aiflow/notification-service/get_latest_v" +
+      "ersion_by_key:\001*B<\n\035com.aiflow.notificat" +
+      "ion.protoZ\025/notification_service\210\001\001\220\001\001b\006" +
+      "proto3"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -14237,7 +14429,7 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.GetLatestVers
     internal_static_notification_service_EventProto_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_notification_service_EventProto_descriptor,
-        new java.lang.String[] { "Key", "Value", "EventType", "Context", "Namespace", "Version", "CreateTime", });
+        new java.lang.String[] { "Key", "Value", "EventType", "Context", "Namespace", "Version", "CreateTime", "Sender", });
     internal_static_notification_service_SendEventRequest_descriptor =
       getDescriptor().getMessageTypes().get(1);
     internal_static_notification_service_SendEventRequest_fieldAccessorTable = new
@@ -14255,7 +14447,7 @@ public com.aiflow.notification.proto.NotificationServiceOuterClass.GetLatestVers
     internal_static_notification_service_ListEventsRequest_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_notification_service_ListEventsRequest_descriptor,
-        new java.lang.String[] { "Keys", "EventType", "StartTime", "StartVersion", "TimeoutSeconds", "Namespace", });
+        new java.lang.String[] { "Keys", "EventType", "StartTime", "StartVersion", "TimeoutSeconds", "Namespace", "Sender", });
     internal_static_notification_service_ListAllEventsRequest_descriptor =
       getDescriptor().getMessageTypes().get(4);
     internal_static_notification_service_ListAllEventsRequest_fieldAccessorTable = new
diff --git a/flink-ai-flow/lib/notification_service/notification_service/base_notification.py b/flink-ai-flow/lib/notification_service/notification_service/base_notification.py
index b42ef0519..19bab6127 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/base_notification.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/base_notification.py
@@ -25,6 +25,7 @@
 ANY_CONDITION = "*"
 DEFAULT_NAMESPACE = "default"
 
+
 class BaseEvent(object):
     def __init__(self,
                  key: str,
@@ -33,7 +34,8 @@ def __init__(self,
                  version: int = None,
                  create_time: int = None,
                  context: str = None,
-                 namespace: str = None):
+                 namespace: str = None,
+                 sender: str = None):
         self.key = key
         self.value = value
         self.event_type = event_type
@@ -41,12 +43,13 @@ def __init__(self,
         self.create_time = create_time
         self.context = context
         self.namespace = namespace
+        self.sender = sender
 
     def __str__(self) -> str:
         return 'key:{0}, value:{1}, type:{2}, version:{3}, create_time:{4}, ' \
-               'context: {5}, namespace: {6}' \
+               'context: {5}, namespace: {6}, sender: {7}' \
             .format(self.key, self.value, self.event_type, self.version, self.create_time,
-                    self.context, self.namespace)
+                    self.context, self.namespace, self.sender)
 
     def __eq__(self, other):
         if not isinstance(other, BaseEvent):
@@ -57,7 +60,8 @@ def __eq__(self, other):
             self.version == other.version and \
             self.create_time == other.create_time and \
             self.context == other.context and \
-            self.namespace == other.namespace
+            self.namespace == other.namespace and \
+            self.sender == other.sender
 
 
 class EventWatcher(metaclass=abc.ABCMeta):
@@ -96,16 +100,20 @@ def send_event(self, event: BaseEvent):
     @abc.abstractmethod
     def list_events(self,
                     key: Union[str, List[str]],
+                    namespace: str = None,
                     version: int = None,
                     event_type: str = None,
-                    start_time: int = None) -> List[BaseEvent]:
+                    start_time: int = None,
+                    sender: str = None) -> List[BaseEvent]:
         """
         List specific events in Notification Service.
 
         :param key: Key of the event for listening.
+        :param namespace: Namespace of the event for listening.
         :param version: (Optional) The version of the events must greater than this version.
         :param event_type: (Optional) Type of the events.
         :param start_time: (Optional) Start time of the events.
+        :param sender: The event sender.
         :return: The event list.
         """
         pass
@@ -114,27 +122,38 @@ def list_events(self,
     def start_listen_event(self,
                            key: Union[str, Tuple[str]],
                            watcher: EventWatcher,
+                           namespace: str = None,
                            version: int = None,
                            event_type: str = None,
-                           start_time: int = None) -> EventWatcherHandle:
+                           start_time: int = None,
+                           sender: str = None) -> EventWatcherHandle:
         """
         Start listen specific `key` or `version` notifications in Notification Service.
 
         :param key: Key of notification for listening.
+        :param namespace: Namespace of the event for listening.
         :param watcher: Watcher instance for listening.
         :param version: (Optional) The version of the events must greater than this version.
         :param event_type: (Optional) Type of the events for listening.
         :param start_time: (Optional) Start time of the events for listening.
+        :param sender: The event sender.
         :return: The handle used to stop the listening.
         """
         pass
 
     @abc.abstractmethod
-    def stop_listen_event(self, key: Union[str, Tuple[str]] = None):
+    def stop_listen_event(self, key: Union[str, Tuple[str]] = None,
+                          namespace: str = None,
+                          event_type: str = None,
+                          sender: str = None
+                          ):
         """
         Stop listen specific `key` notifications in Notification Service.
 
         :param key: Keys of notification for listening.
+        :param namespace: Namespace of notification for listening.
+        :param event_type: (Optional) Type of the events for listening.
+        :param sender: The event sender.
         """
         pass
 
@@ -178,10 +197,11 @@ def stop_listen_events(self):
         pass
 
     @abc.abstractmethod
-    def get_latest_version(self, key: str = None):
+    def get_latest_version(self, key: str = None, namespace: str = None):
         """
         get latest event's version by key.
         :param key: Key of notification for listening.
+        :param namespace: Namespace of notification for listening.
         :return: Version number of the specific key.
         """
         pass
diff --git a/flink-ai-flow/lib/notification_service/notification_service/client.py b/flink-ai-flow/lib/notification_service/notification_service/client.py
index ec71833c0..dec599525 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/client.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/client.py
@@ -17,8 +17,9 @@
 # under the License.
 #
 import logging
-import threading
 import os
+import threading
+import time
 import uuid
 from collections import Iterable
 from functools import wraps
@@ -26,13 +27,13 @@
 from typing import Union, List, Tuple, Dict, Any
 
 import grpc
-import time
 
-from notification_service.base_notification import BaseNotification, EventWatcher, BaseEvent, EventWatcherHandle
+from notification_service.base_notification import BaseNotification, EventWatcher, BaseEvent, EventWatcherHandle, \
+    ANY_CONDITION
+from notification_service.proto import notification_service_pb2_grpc
 from notification_service.proto.notification_service_pb2 \
     import SendEventRequest, ListEventsRequest, EventProto, ReturnStatus, ListAllEventsRequest, \
     GetLatestVersionByKeyRequest, ListMembersRequest
-from notification_service.proto import notification_service_pb2_grpc
 from notification_service.util.utils import event_proto_to_event, proto_to_member, sleep_and_detecting_running
 
 if not hasattr(time, 'time_ns'):
@@ -73,7 +74,8 @@ def __init__(self,
                  enable_ha: bool = False,
                  list_member_interval_ms: int = 5000,
                  retry_interval_ms: int = 1000,
-                 retry_timeout_ms: int = 10000):
+                 retry_timeout_ms: int = 10000,
+                 sender: str = None):
         """
         The constructor of the NotificationClient.
 
@@ -92,6 +94,7 @@ def __init__(self,
         :param retry_timeout_ms: When `enable_ha` is True and a rpc call has failed on all the
                                  living members, this client will retry until success or timeout.
                                  This param specifies the retry timeout.
+        :param sender: The identify of the client.
         """
         channel = grpc.insecure_channel(server_uri)
         self._default_namespace = default_namespace
@@ -102,6 +105,7 @@ def __init__(self,
         self.list_member_interval_ms = list_member_interval_ms
         self.retry_interval_ms = retry_interval_ms
         self.retry_timeout_ms = retry_timeout_ms
+        self._sender = sender
         if self.enable_ha:
             server_uris = server_uri.split(",")
             self.living_members = []
@@ -144,7 +148,9 @@ def send_event(self, event: BaseEvent):
                 value=event.value,
                 event_type=event.event_type,
                 context=event.context,
-                namespace=self._default_namespace if event.namespace is None else event.namespace),
+                namespace=self._default_namespace,
+                sender=self._sender
+            ),
             uuid=str(uuid.uuid4()))
         response = self.notification_stub.sendEvent(request)
         if response.return_code == ReturnStatus.SUCCESS:
@@ -157,15 +163,17 @@ def list_events(self,
                     namespace: str = None,
                     version: int = None,
                     event_type: str = None,
-                    start_time: int = None) -> List[BaseEvent]:
+                    start_time: int = None,
+                    sender: str = None) -> List[BaseEvent]:
         """
         List specific events in Notification Service.
 
         :param key: Key of the event for listening.
-        :param namespace: Namespace of the event for listening.
-        :param version: (Optional) The version of the events must greater than this version.
+        :param namespace: (Optional) Namespace of the event for listening.
+        :param version: (Optional) Version of the events must greater than this version.
         :param event_type: (Optional) Type of the events.
         :param start_time: (Optional) Start time of the events.
+        :param sender: The event sender.
         :return: The event list.
         """
         if isinstance(key, str):
@@ -177,7 +185,9 @@ def list_events(self,
             start_version=version,
             event_type=event_type,
             start_time=start_time,
-            namespace=self._default_namespace if namespace is None else namespace)
+            namespace=self._default_namespace if namespace is None else namespace,
+            sender=sender
+        )
         response = self.notification_stub.listEvents(request)
         if response.return_code == ReturnStatus.SUCCESS:
             if response.events is None:
@@ -197,16 +207,18 @@ def start_listen_event(self,
                            namespace: str = None,
                            version: int = None,
                            event_type: str = None,
-                           start_time: int = None) -> EventWatcherHandle:
+                           start_time: int = None,
+                           sender: str = None) -> EventWatcherHandle:
         """
         Start listen specific `key` or `version` notifications in Notification Service.
 
         :param key: Key of notification for listening.
         :param watcher: Watcher instance for listening.
-        :param namespace: Namespace of the event for listening.
+        :param namespace: (Optional) Namespace of the event for listening.
         :param version: (Optional) The version of the events must greater than this version.
         :param event_type: (Optional) Type of the events for listening.
         :param start_time: (Optional) Start time of the events for listening.
+        :param sender: The event sender.
         :return: The handle used to stop the listening.
         """
         if isinstance(key, str):
@@ -214,6 +226,8 @@ def start_listen_event(self,
         elif isinstance(key, Iterable):
             key = tuple(key)
         namespace = self._default_namespace if namespace is None else namespace
+        sender = ANY_CONDITION if sender is None else sender
+        event_type = ANY_CONDITION if event_type is None else event_type
 
         def list_events(client,
                         k: Tuple[str],
@@ -221,6 +235,7 @@ def list_events(client,
                         t: str = None,
                         ts: int = None,
                         ns: str = None,
+                        sd: str = None,
                         timeout_seconds: int = None):
             request = ListEventsRequest(
                 keys=k,
@@ -228,7 +243,9 @@ def list_events(client,
                 start_time=ts,
                 start_version=v,
                 timeout_seconds=timeout_seconds,
-                namespace=ns)
+                namespace=ns,
+                sender=sd
+            )
             response = client.notification_stub.listEvents(request)
             if response.return_code == ReturnStatus.SUCCESS:
                 if response.events is None:
@@ -242,7 +259,7 @@ def list_events(client,
             else:
                 raise Exception(response.return_msg)
 
-        def listen(client, k, v, t, ts, ns, w):
+        def listen(client, k, v, t, ts, ns, sd, w):
             th = threading.current_thread()
             listen_version = v
             while getattr(th, '_flag', True):
@@ -253,6 +270,7 @@ def listen(client, k, v, t, ts, ns, w):
                     t,
                     ts,
                     ns,
+                    sd,
                     NOTIFICATION_TIMEOUT_SECONDS)
                 if len(notifications) > 0:
                     w.process(notifications)
@@ -260,27 +278,34 @@ def listen(client, k, v, t, ts, ns, w):
 
         thread = threading.Thread(
             target=listen,
-            args=(self, key, version, event_type, start_time, namespace,
+            args=(self, key, version, event_type, start_time, namespace, sender,
                   watcher),
             daemon=True)
         thread.start()
         self.lock.acquire()
         try:
-            if self.threads.get((key, namespace)) is None:
-                self.threads[(key, namespace)] = []
-            self.threads[(key, namespace)].append(thread)
+            if self.threads.get((key, namespace, event_type, sender)) is None:
+                self.threads[(key, namespace, event_type, sender)] = []
+            self.threads[(key, namespace, event_type, sender)].append(thread)
         finally:
             self.lock.release()
-        return ThreadEventWatcherHandle(thread, (key, namespace), self)
+        return ThreadEventWatcherHandle(thread, (key, namespace, event_type, sender), self)
 
-    def stop_listen_event(self, key: Union[str, Tuple[str]] = None, namespace: str = None):
+    def stop_listen_event(self, key: Union[str, Tuple[str]] = None,
+                          namespace: str = None,
+                          event_type: str = None,
+                          sender: str = None):
         """
         Stop listen specific `key` notifications in Notification Service.
 
         :param key: Keys of notification for listening.
-        :param namespace: Namespace of notification for listening.
+        :param namespace: (Optional) Namespace of notification for listening.
+        :param event_type: (Optional) Type of the events for listening.
+        :param sender: The event sender.
         """
         namespace = self._default_namespace if namespace is None else namespace
+        event_type = ANY_CONDITION if event_type is None else event_type
+        sender = ANY_CONDITION if sender is None else sender
         if key is None:
             for (thread_key, v) in self.threads.items():
                 if thread_key == ALL_EVENTS_KEY:
@@ -297,12 +322,12 @@ def stop_listen_event(self, key: Union[str, Tuple[str]] = None, namespace: str =
             if isinstance(key, str):
                 key = (key,)
             try:
-                if (key, namespace) in self.threads:
-                    threads = self.threads[(key, namespace)]
+                if (key, namespace, event_type, sender) in self.threads:
+                    threads = self.threads[(key, namespace, event_type, sender)]
                     for thread in threads:
                         thread._flag = False
                         thread.join()
-                    del self.threads[(key, namespace)]
+                    del self.threads[(key, namespace, event_type, sender)]
             finally:
                 self.lock.release()
 
@@ -431,8 +456,8 @@ def stop_listen_events(self):
     def get_latest_version(self, key: str = None, namespace: str = None):
         """
         get latest event's version by key.
-        :param key: Key of notification for listening.
-        :param namespace: Namespace of notification for listening.
+        :param key: (Optional) Key of notification for listening.
+        :param namespace: (Optional) Namespace of notification for listening.
         :return: Version number of the specific key.
         """
         self.lock.acquire()
@@ -452,10 +477,6 @@ def disable_high_availability(self):
 
     def _list_members(self):
         while self.ha_running:
-            # sleep for a `list_member_interval`
-            # sleep_and_detecting_running(self.list_member_interval_ms,
-            #                             lambda: self.ha_running)
-
             # refresh the living members
             request = ListMembersRequest(timeout_seconds=int(self.list_member_interval_ms / 1000))
             response = self.notification_stub.listMembers(request)
diff --git a/flink-ai-flow/lib/notification_service/notification_service/event_storage.py b/flink-ai-flow/lib/notification_service/notification_service/event_storage.py
index c69b17ba5..bdb81b009 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/event_storage.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/event_storage.py
@@ -38,7 +38,8 @@ def list_events(self,
                     version: int = None,
                     event_type: str = None,
                     start_time: int = None,
-                    namespace: str = None):
+                    namespace: str = None,
+                    sender: str = None):
         pass
 
     @abstractmethod
@@ -66,7 +67,7 @@ def __init__(self):
 
     def add_event(self, event: BaseEvent, uuid: str):
         self.max_version += 1
-        event.create_time = time.time_ns()
+        event.create_time = int(time.time() * 1000)
         event.version = self.max_version
         self.store.append(event)
         return event
@@ -76,12 +77,14 @@ def list_events(self,
                     version: int = None,
                     event_type: str = None,
                     start_time: int = None,
-                    namespace: str = None):
+                    namespace: str = None,
+                    sender: str = None):
         res = []
         key = None if key == "" else key
         version = None if version == 0 else version
         event_type = None if event_type == "" else event_type
         namespace = None if namespace == "" else namespace
+        sender = None if sender == "" else sender
         if isinstance(key, str):
             key = (key, )
         elif isinstance(key, Iterable):
@@ -95,7 +98,9 @@ def list_events(self,
                 continue
             if start_time is not None and event.create_time < start_time:
                 continue
-            if namespace != ANY_CONDITION and event.namespace != namespace:
+            if namespace is not None and namespace != ANY_CONDITION and event.namespace != namespace:
+                continue
+            if sender is not None and sender != ANY_CONDITION and event.sender != sender:
                 continue
             res.append(event)
         return res
@@ -140,8 +145,9 @@ def list_events(self,
                     version: int = None,
                     event_type: str = None,
                     start_time: int = None,
-                    namespace: str = None):
-        return EventModel.list_events(key, version, event_type, start_time, namespace)
+                    namespace: str = None,
+                    sender: str = None):
+        return EventModel.list_events(key, version, event_type, start_time, namespace, sender)
 
     def list_all_events(self, start_time: int):
         return EventModel.list_all_events(start_time)
diff --git a/flink-ai-flow/lib/notification_service/notification_service/mongo_event_storage.py b/flink-ai-flow/lib/notification_service/notification_service/mongo_event_storage.py
index b0e71ed23..6acc395ad 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/mongo_event_storage.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/mongo_event_storage.py
@@ -66,6 +66,7 @@ def add_event(self, event: BaseEvent, uuid: str):
             "value": event.value,
             "context": event.context,
             "namespace": event.namespace,
+            "sender": event.sender,
             "uuid": uuid
         }
         mongo_event = MongoEvent(**kwargs)
@@ -80,16 +81,18 @@ def list_events(self,
                     version: int = None,
                     event_type: str = None,
                     start_time: int = None,
-                    namespace: str = None):
+                    namespace: str = None,
+                    sender: str = None):
         key = None if key == "" else key
         version = None if version == 0 else version
         event_type = None if event_type == "" else event_type
         namespace = None if namespace == "" else namespace
+        sender = None if sender == "" else sender
         if isinstance(key, str):
             key = (key,)
         elif isinstance(key, Iterable):
             key = tuple(key)
-        res = MongoEvent.get_base_events(key, version, event_type, start_time, namespace)
+        res = MongoEvent.get_base_events(key, version, event_type, start_time, namespace, sender)
         return res
 
     def list_all_events(self, start_time: int):
diff --git a/flink-ai-flow/lib/notification_service/notification_service/mongo_notification.py b/flink-ai-flow/lib/notification_service/notification_service/mongo_notification.py
index a2ceffa14..023338b66 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/mongo_notification.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/mongo_notification.py
@@ -31,6 +31,7 @@ class MongoEvent(Document):
     create_time = IntField()
     context = StringField()
     namespace = StringField()
+    sender = StringField()
     uuid = StringField()
 
     def to_dict(self):
@@ -41,7 +42,8 @@ def to_dict(self):
             "version": int(self.version),
             "create_time": self.create_time,
             "context": self.context,
-            "namespace": self.namespace
+            "namespace": self.namespace,
+            "sender": self.sender
         }
 
     def to_base_event(self):
@@ -72,7 +74,8 @@ def get_base_events(cls,
                         version: int = None,
                         event_type: str = None,
                         start_time: int = None,
-                        namespace: str = None):
+                        namespace: str = None,
+                        sender: str = None):
         conditions = dict()
         if len(key) == 1:
             if ANY_CONDITION != key[0]:
@@ -88,6 +91,8 @@ def get_base_events(cls,
             conditions["start_time_gte"] = start_time
         if ANY_CONDITION != namespace:
             conditions["namespace"] = namespace
+        if sender is not None and ANY_CONDITION != sender:
+            conditions["sender"] = sender
         mongo_events = cls.objects(**conditions).order_by("version")
         return cls.convert_to_base_events(mongo_events)
 
diff --git a/flink-ai-flow/lib/notification_service/notification_service/proto/gen_protobuf.sh b/flink-ai-flow/lib/notification_service/notification_service/proto/gen_protobuf.sh
index 9e85620b4..0e572c714 100755
--- a/flink-ai-flow/lib/notification_service/notification_service/proto/gen_protobuf.sh
+++ b/flink-ai-flow/lib/notification_service/notification_service/proto/gen_protobuf.sh
@@ -58,6 +58,6 @@ rm -rf *.py-E
 
 cd ..
 
-sed -i -E 's/^package notification_service/package service/' go/*.go
-sed -i '' 's/\_ "github.com\/grpc-ecosystem\/grpc-gateway\/third_party\/googleapis\/google\/api"/\/\/\_ "github.com\/grpc-ecosystem\/grpc-gateway\/third_party\/googleapis\/google\/api"/g' go/notification_service.pb.go
+sed -i -E 's/^package notification_service/package service/' go/notification_service/*.go
+sed -i '' 's/\_ "github.com\/grpc-ecosystem\/grpc-gateway\/third_party\/googleapis\/google\/api"/\/\/\_ "github.com\/grpc-ecosystem\/grpc-gateway\/third_party\/googleapis\/google\/api"/g' go/notification_service/notification_service.pb.go
 rm -rf go/*.go-E
diff --git a/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service.proto b/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service.proto
index b05475e0c..a09d3616d 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service.proto
+++ b/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service.proto
@@ -19,6 +19,7 @@
 syntax = "proto3";
 option java_generic_services = true;
 option java_package = "com.aiflow.notification.proto";
+option go_package = "/notification_service";
 option py_generic_services = true;
 
 import "google/api/annotations.proto";
@@ -95,6 +96,7 @@ message EventProto {
     string namespace = 5;
     int64 version = 6;
     int64 create_time = 7;
+    string sender = 8;
 }
 
 
@@ -122,6 +124,7 @@ message ListEventsRequest {
     int64 start_version = 4;
     int32 timeout_seconds = 5;
     string namespace = 6;
+    string sender = 7;
 }
 
 message ListAllEventsRequest {
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 a14b809f5..c00b0fcda 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
@@ -19,7 +19,9 @@
 #
 # Generated by the protocol buffer compiler.  DO NOT EDIT!
 # source: notification_service.proto
-"""Generated protocol buffer code."""
+
+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
@@ -27,6 +29,7 @@
 from google.protobuf import symbol_database as _symbol_database
 from google.protobuf import service as _service
 from google.protobuf import service_reflection
+from google.protobuf import descriptor_pb2
 # @@protoc_insertion_point(imports)
 
 _sym_db = _symbol_database.Default()
@@ -39,9 +42,7 @@
   name='notification_service.proto',
   package='notification_service',
   syntax='proto3',
-  serialized_options=b'\n\035com.aiflow.notification.proto\210\001\001\220\001\001',
-  create_key=_descriptor._internal_create_key,
-  serialized_pb=b'\n\x1anotification_service.proto\x12\x14notification_service\x1a\x1cgoogle/api/annotations.proto\"\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\xef\x08\n\x13NotificationService\x12\x91\x01\n\tsendEvent\x12&.notification_service.SendEventRequest\x1a(.notification_service.SendEventsResponse\"2\x82\xd3\xe4\x93\x02,\"\'/aiflow/notification-service/send_event:\x01*\x12\x94\x01\n\nlistEvents\x12\'.notification_service.ListEventsRequest\x1a(.notification_service.ListEventsResponse\"3\x82\xd3\xe4\x93\x02-\"(/aiflow/notification-service/list_events:\x01*\x12\x9e\x01\n\rlistAllEvents\x12*.notification_service.ListAllEventsRequest\x1a(.notification_service.ListEventsResponse\"7\x82\xd3\xe4\x93\x02\x31\",/aiflow/notification-service/list_all_events:\x01*\x12\x83\x01\n\x06notify\x12#.notification_service.NotifyRequest\x1a$.notification_service.NotifyResponse\".\x82\xd3\xe4\x93\x02(\"#/aiflow/notification-service/notify:\x01*\x12\x98\x01\n\x0blistMembers\x12(.notification_service.ListMembersRequest\x1a).notification_service.ListMembersResponse\"4\x82\xd3\xe4\x93\x02.\")/aiflow/notification-service/list_members:\x01*\x12\xa9\x01\n\x0fnotifyNewMember\x12,.notification_service.NotifyNewMemberRequest\x1a-.notification_service.NotifyNewMemberResponse\"9\x82\xd3\xe4\x93\x02\x33\"./aiflow/notification-service/notify_new_member:\x01*\x12\xbe\x01\n\x15getLatestVersionByKey\x12\x32.notification_service.GetLatestVersionByKeyRequest\x1a..notification_service.GetLatestVersionResponse\"A\x82\xd3\xe4\x93\x02;\"6/aiflow/notification-service/get_latest_version_by_key:\x01*B%\n\x1d\x63om.aiflow.notification.proto\x88\x01\x01\x90\x01\x01\x62\x06proto3'
+  serialized_pb=_b('\n\x1anotification_service.proto\x12\x14notification_service\x1a\x1cgoogle/api/annotations.proto\"\x96\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\x12\x0e\n\x06sender\x18\x08 \x01(\t\"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\"\x9c\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\x12\x0e\n\x06sender\x18\x07 \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\xef\x08\n\x13NotificationService\x12\x91\x01\n\tsendEvent\x12&.notification_service.SendEventRequest\x1a(.notification_service.SendEventsResponse\"2\x82\xd3\xe4\x93\x02,\"\'/aiflow/notification-service/send_event:\x01*\x12\x94\x01\n\nlistEvents\x12\'.notification_service.ListEventsRequest\x1a(.notification_service.ListEventsResponse\"3\x82\xd3\xe4\x93\x02-\"(/aiflow/notification-service/list_events:\x01*\x12\x9e\x01\n\rlistAllEvents\x12*.notification_service.ListAllEventsRequest\x1a(.notification_service.ListEventsResponse\"7\x82\xd3\xe4\x93\x02\x31\",/aiflow/notification-service/list_all_events:\x01*\x12\x83\x01\n\x06notify\x12#.notification_service.NotifyRequest\x1a$.notification_service.NotifyResponse\".\x82\xd3\xe4\x93\x02(\"#/aiflow/notification-service/notify:\x01*\x12\x98\x01\n\x0blistMembers\x12(.notification_service.ListMembersRequest\x1a).notification_service.ListMembersResponse\"4\x82\xd3\xe4\x93\x02.\")/aiflow/notification-service/list_members:\x01*\x12\xa9\x01\n\x0fnotifyNewMember\x12,.notification_service.NotifyNewMemberRequest\x1a-.notification_service.NotifyNewMemberResponse\"9\x82\xd3\xe4\x93\x02\x33\"./aiflow/notification-service/notify_new_member:\x01*\x12\xbe\x01\n\x15getLatestVersionByKey\x12\x32.notification_service.GetLatestVersionByKeyRequest\x1a..notification_service.GetLatestVersionResponse\"A\x82\xd3\xe4\x93\x02;\"6/aiflow/notification-service/get_latest_version_by_key:\x01*B<\n\x1d\x63om.aiflow.notification.protoZ\x15/notification_service\x88\x01\x01\x90\x01\x01\x62\x06proto3')
   ,
   dependencies=[google_dot_api_dot_annotations__pb2.DESCRIPTOR,])
 
@@ -50,23 +51,20 @@
   full_name='notification_service.ReturnStatus',
   filename=None,
   file=DESCRIPTOR,
-  create_key=_descriptor._internal_create_key,
   values=[
     _descriptor.EnumValueDescriptor(
       name='SUCCESS', index=0, number=0,
-      serialized_options=None,
-      type=None,
-      create_key=_descriptor._internal_create_key),
+      options=None,
+      type=None),
     _descriptor.EnumValueDescriptor(
       name='ERROR', index=1, number=1,
-      serialized_options=None,
-      type=None,
-      create_key=_descriptor._internal_create_key),
+      options=None,
+      type=None),
   ],
   containing_type=None,
-  serialized_options=None,
-  serialized_start=1688,
-  serialized_end=1726,
+  options=None,
+  serialized_start=1720,
+  serialized_end=1758,
 )
 _sym_db.RegisterEnumDescriptor(_RETURNSTATUS)
 
@@ -82,71 +80,77 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _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,  create_key=_descriptor._internal_create_key),
+      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,  create_key=_descriptor._internal_create_key),
+      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,  create_key=_descriptor._internal_create_key),
+      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,  create_key=_descriptor._internal_create_key),
+      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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='version', full_name='notification_service.EventProto.version', index=5,
       number=6, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='create_time', full_name='notification_service.EventProto.create_time', index=6,
       number=7, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
+    _descriptor.FieldDescriptor(
+      name='sender', full_name='notification_service.EventProto.sender', index=7,
+      number=8, type=9, cpp_type=9, label=1,
+      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,
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
   serialized_start=83,
-  serialized_end=217,
+  serialized_end=233,
 )
 
 
@@ -156,7 +160,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='event', full_name='notification_service.SendEventRequest.event', index=0,
@@ -164,28 +167,28 @@
       has_default_value=False, default_value=None,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=219,
-  serialized_end=300,
+  serialized_start=235,
+  serialized_end=316,
 )
 
 
@@ -195,7 +198,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='event', full_name='notification_service.SendEventsResponse.event', index=0,
@@ -203,35 +205,35 @@
       has_default_value=False, default_value=None,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='return_code', full_name='notification_service.SendEventsResponse.return_code', index=1,
       number=2, type=14, cpp_type=8, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=303,
-  serialized_end=449,
+  serialized_start=319,
+  serialized_end=465,
 )
 
 
@@ -241,7 +243,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='keys', full_name='notification_service.ListEventsRequest.keys', index=0,
@@ -249,56 +250,63 @@
       has_default_value=False, default_value=[],
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='start_time', full_name='notification_service.ListEventsRequest.start_time', index=2,
       number=3, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='start_version', full_name='notification_service.ListEventsRequest.start_version', index=3,
       number=4, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='timeout_seconds', full_name='notification_service.ListEventsRequest.timeout_seconds', index=4,
       number=5, type=5, cpp_type=1, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,
+      options=None, file=DESCRIPTOR),
+    _descriptor.FieldDescriptor(
+      name='sender', full_name='notification_service.ListEventsRequest.sender', index=6,
+      number=7, type=9, cpp_type=9, label=1,
+      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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=452,
-  serialized_end=592,
+  serialized_start=468,
+  serialized_end=624,
 )
 
 
@@ -308,7 +316,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='timeout_seconds', full_name='notification_service.ListAllEventsRequest.timeout_seconds', index=0,
@@ -316,42 +323,42 @@
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='start_time', full_name='notification_service.ListAllEventsRequest.start_time', index=1,
       number=2, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='start_version', full_name='notification_service.ListAllEventsRequest.start_version', index=2,
       number=3, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='end_version', full_name='notification_service.ListAllEventsRequest.end_version', index=3,
       number=4, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=594,
-  serialized_end=705,
+  serialized_start=626,
+  serialized_end=737,
 )
 
 
@@ -361,7 +368,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='return_code', full_name='notification_service.ListEventsResponse.return_code', index=0,
@@ -369,35 +375,35 @@
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='events', full_name='notification_service.ListEventsResponse.events', index=2,
       number=3, type=11, cpp_type=10, label=3,
       has_default_value=False, default_value=[],
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=708,
-  serialized_end=855,
+  serialized_start=740,
+  serialized_end=887,
 )
 
 
@@ -407,43 +413,42 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _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,  create_key=_descriptor._internal_create_key),
+      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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='timeout_seconds', full_name='notification_service.GetLatestVersionByKeyRequest.timeout_seconds', index=2,
       number=3, type=5, cpp_type=1, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=857,
-  serialized_end=944,
+  serialized_start=889,
+  serialized_end=976,
 )
 
 
@@ -453,43 +458,42 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _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,  create_key=_descriptor._internal_create_key),
+      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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='version', full_name='notification_service.GetLatestVersionResponse.version', index=2,
       number=3, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=946,
-  serialized_end=1030,
+  serialized_start=978,
+  serialized_end=1062,
 )
 
 
@@ -499,7 +503,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='version', full_name='notification_service.MemberProto.version', index=0,
@@ -507,35 +510,35 @@
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='update_time', full_name='notification_service.MemberProto.update_time', index=2,
       number=3, type=3, cpp_type=2, label=1,
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1032,
-  serialized_end=1103,
+  serialized_start=1064,
+  serialized_end=1135,
 )
 
 
@@ -545,36 +548,35 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _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,  create_key=_descriptor._internal_create_key),
+      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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1105,
-  serialized_end=1145,
+  serialized_start=1137,
+  serialized_end=1177,
 )
 
 
@@ -584,7 +586,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='notifies', full_name='notification_service.NotifyRequest.notifies', index=0,
@@ -592,21 +593,21 @@
       has_default_value=False, default_value=[],
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1147,
-  serialized_end=1210,
+  serialized_start=1179,
+  serialized_end=1242,
 )
 
 
@@ -616,7 +617,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='return_code', full_name='notification_service.NotifyResponse.return_code', index=0,
@@ -624,28 +624,28 @@
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1212,
-  serialized_end=1305,
+  serialized_start=1244,
+  serialized_end=1337,
 )
 
 
@@ -655,7 +655,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='timeout_seconds', full_name='notification_service.ListMembersRequest.timeout_seconds', index=0,
@@ -663,21 +662,21 @@
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1307,
-  serialized_end=1352,
+  serialized_start=1339,
+  serialized_end=1384,
 )
 
 
@@ -687,7 +686,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='return_code', full_name='notification_service.ListMembersResponse.return_code', index=0,
@@ -695,35 +693,35 @@
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _descriptor.FieldDescriptor(
       name='members', full_name='notification_service.ListMembersResponse.members', index=2,
       number=3, type=11, cpp_type=10, label=3,
       has_default_value=False, default_value=[],
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1355,
-  serialized_end=1505,
+  serialized_start=1387,
+  serialized_end=1537,
 )
 
 
@@ -733,7 +731,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='member', full_name='notification_service.NotifyNewMemberRequest.member', index=0,
@@ -741,21 +738,21 @@
       has_default_value=False, default_value=None,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1507,
-  serialized_end=1582,
+  serialized_start=1539,
+  serialized_end=1614,
 )
 
 
@@ -765,7 +762,6 @@
   filename=None,
   file=DESCRIPTOR,
   containing_type=None,
-  create_key=_descriptor._internal_create_key,
   fields=[
     _descriptor.FieldDescriptor(
       name='return_code', full_name='notification_service.NotifyNewMemberResponse.return_code', index=0,
@@ -773,28 +769,28 @@
       has_default_value=False, default_value=0,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
-      serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
     _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,  create_key=_descriptor._internal_create_key),
+      options=None, file=DESCRIPTOR),
   ],
   extensions=[
   ],
   nested_types=[],
   enum_types=[
   ],
-  serialized_options=None,
+  options=None,
   is_extendable=False,
   syntax='proto3',
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1584,
-  serialized_end=1686,
+  serialized_start=1616,
+  serialized_end=1718,
 )
 
 _SENDEVENTREQUEST.fields_by_name['event'].message_type = _EVENTPROTO
@@ -827,130 +823,130 @@
 DESCRIPTOR.enum_types_by_name['ReturnStatus'] = _RETURNSTATUS
 _sym_db.RegisterFileDescriptor(DESCRIPTOR)
 
-EventProto = _reflection.GeneratedProtocolMessageType('EventProto', (_message.Message,), {
-  'DESCRIPTOR' : _EVENTPROTO,
-  '__module__' : 'notification_service_pb2'
+EventProto = _reflection.GeneratedProtocolMessageType('EventProto', (_message.Message,), dict(
+  DESCRIPTOR = _EVENTPROTO,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.EventProto)
-  })
+  ))
 _sym_db.RegisterMessage(EventProto)
 
-SendEventRequest = _reflection.GeneratedProtocolMessageType('SendEventRequest', (_message.Message,), {
-  'DESCRIPTOR' : _SENDEVENTREQUEST,
-  '__module__' : 'notification_service_pb2'
+SendEventRequest = _reflection.GeneratedProtocolMessageType('SendEventRequest', (_message.Message,), dict(
+  DESCRIPTOR = _SENDEVENTREQUEST,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.SendEventRequest)
-  })
+  ))
 _sym_db.RegisterMessage(SendEventRequest)
 
-SendEventsResponse = _reflection.GeneratedProtocolMessageType('SendEventsResponse', (_message.Message,), {
-  'DESCRIPTOR' : _SENDEVENTSRESPONSE,
-  '__module__' : 'notification_service_pb2'
+SendEventsResponse = _reflection.GeneratedProtocolMessageType('SendEventsResponse', (_message.Message,), dict(
+  DESCRIPTOR = _SENDEVENTSRESPONSE,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.SendEventsResponse)
-  })
+  ))
 _sym_db.RegisterMessage(SendEventsResponse)
 
-ListEventsRequest = _reflection.GeneratedProtocolMessageType('ListEventsRequest', (_message.Message,), {
-  'DESCRIPTOR' : _LISTEVENTSREQUEST,
-  '__module__' : 'notification_service_pb2'
+ListEventsRequest = _reflection.GeneratedProtocolMessageType('ListEventsRequest', (_message.Message,), dict(
+  DESCRIPTOR = _LISTEVENTSREQUEST,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.ListEventsRequest)
-  })
+  ))
 _sym_db.RegisterMessage(ListEventsRequest)
 
-ListAllEventsRequest = _reflection.GeneratedProtocolMessageType('ListAllEventsRequest', (_message.Message,), {
-  'DESCRIPTOR' : _LISTALLEVENTSREQUEST,
-  '__module__' : 'notification_service_pb2'
+ListAllEventsRequest = _reflection.GeneratedProtocolMessageType('ListAllEventsRequest', (_message.Message,), dict(
+  DESCRIPTOR = _LISTALLEVENTSREQUEST,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.ListAllEventsRequest)
-  })
+  ))
 _sym_db.RegisterMessage(ListAllEventsRequest)
 
-ListEventsResponse = _reflection.GeneratedProtocolMessageType('ListEventsResponse', (_message.Message,), {
-  'DESCRIPTOR' : _LISTEVENTSRESPONSE,
-  '__module__' : 'notification_service_pb2'
+ListEventsResponse = _reflection.GeneratedProtocolMessageType('ListEventsResponse', (_message.Message,), dict(
+  DESCRIPTOR = _LISTEVENTSRESPONSE,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.ListEventsResponse)
-  })
+  ))
 _sym_db.RegisterMessage(ListEventsResponse)
 
-GetLatestVersionByKeyRequest = _reflection.GeneratedProtocolMessageType('GetLatestVersionByKeyRequest', (_message.Message,), {
-  'DESCRIPTOR' : _GETLATESTVERSIONBYKEYREQUEST,
-  '__module__' : 'notification_service_pb2'
+GetLatestVersionByKeyRequest = _reflection.GeneratedProtocolMessageType('GetLatestVersionByKeyRequest', (_message.Message,), dict(
+  DESCRIPTOR = _GETLATESTVERSIONBYKEYREQUEST,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.GetLatestVersionByKeyRequest)
-  })
+  ))
 _sym_db.RegisterMessage(GetLatestVersionByKeyRequest)
 
-GetLatestVersionResponse = _reflection.GeneratedProtocolMessageType('GetLatestVersionResponse', (_message.Message,), {
-  'DESCRIPTOR' : _GETLATESTVERSIONRESPONSE,
-  '__module__' : 'notification_service_pb2'
+GetLatestVersionResponse = _reflection.GeneratedProtocolMessageType('GetLatestVersionResponse', (_message.Message,), dict(
+  DESCRIPTOR = _GETLATESTVERSIONRESPONSE,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.GetLatestVersionResponse)
-  })
+  ))
 _sym_db.RegisterMessage(GetLatestVersionResponse)
 
-MemberProto = _reflection.GeneratedProtocolMessageType('MemberProto', (_message.Message,), {
-  'DESCRIPTOR' : _MEMBERPROTO,
-  '__module__' : 'notification_service_pb2'
+MemberProto = _reflection.GeneratedProtocolMessageType('MemberProto', (_message.Message,), dict(
+  DESCRIPTOR = _MEMBERPROTO,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.MemberProto)
-  })
+  ))
 _sym_db.RegisterMessage(MemberProto)
 
-Notify = _reflection.GeneratedProtocolMessageType('Notify', (_message.Message,), {
-  'DESCRIPTOR' : _NOTIFY,
-  '__module__' : 'notification_service_pb2'
+Notify = _reflection.GeneratedProtocolMessageType('Notify', (_message.Message,), dict(
+  DESCRIPTOR = _NOTIFY,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.Notify)
-  })
+  ))
 _sym_db.RegisterMessage(Notify)
 
-NotifyRequest = _reflection.GeneratedProtocolMessageType('NotifyRequest', (_message.Message,), {
-  'DESCRIPTOR' : _NOTIFYREQUEST,
-  '__module__' : 'notification_service_pb2'
+NotifyRequest = _reflection.GeneratedProtocolMessageType('NotifyRequest', (_message.Message,), dict(
+  DESCRIPTOR = _NOTIFYREQUEST,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.NotifyRequest)
-  })
+  ))
 _sym_db.RegisterMessage(NotifyRequest)
 
-NotifyResponse = _reflection.GeneratedProtocolMessageType('NotifyResponse', (_message.Message,), {
-  'DESCRIPTOR' : _NOTIFYRESPONSE,
-  '__module__' : 'notification_service_pb2'
+NotifyResponse = _reflection.GeneratedProtocolMessageType('NotifyResponse', (_message.Message,), dict(
+  DESCRIPTOR = _NOTIFYRESPONSE,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.NotifyResponse)
-  })
+  ))
 _sym_db.RegisterMessage(NotifyResponse)
 
-ListMembersRequest = _reflection.GeneratedProtocolMessageType('ListMembersRequest', (_message.Message,), {
-  'DESCRIPTOR' : _LISTMEMBERSREQUEST,
-  '__module__' : 'notification_service_pb2'
+ListMembersRequest = _reflection.GeneratedProtocolMessageType('ListMembersRequest', (_message.Message,), dict(
+  DESCRIPTOR = _LISTMEMBERSREQUEST,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.ListMembersRequest)
-  })
+  ))
 _sym_db.RegisterMessage(ListMembersRequest)
 
-ListMembersResponse = _reflection.GeneratedProtocolMessageType('ListMembersResponse', (_message.Message,), {
-  'DESCRIPTOR' : _LISTMEMBERSRESPONSE,
-  '__module__' : 'notification_service_pb2'
+ListMembersResponse = _reflection.GeneratedProtocolMessageType('ListMembersResponse', (_message.Message,), dict(
+  DESCRIPTOR = _LISTMEMBERSRESPONSE,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.ListMembersResponse)
-  })
+  ))
 _sym_db.RegisterMessage(ListMembersResponse)
 
-NotifyNewMemberRequest = _reflection.GeneratedProtocolMessageType('NotifyNewMemberRequest', (_message.Message,), {
-  'DESCRIPTOR' : _NOTIFYNEWMEMBERREQUEST,
-  '__module__' : 'notification_service_pb2'
+NotifyNewMemberRequest = _reflection.GeneratedProtocolMessageType('NotifyNewMemberRequest', (_message.Message,), dict(
+  DESCRIPTOR = _NOTIFYNEWMEMBERREQUEST,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.NotifyNewMemberRequest)
-  })
+  ))
 _sym_db.RegisterMessage(NotifyNewMemberRequest)
 
-NotifyNewMemberResponse = _reflection.GeneratedProtocolMessageType('NotifyNewMemberResponse', (_message.Message,), {
-  'DESCRIPTOR' : _NOTIFYNEWMEMBERRESPONSE,
-  '__module__' : 'notification_service_pb2'
+NotifyNewMemberResponse = _reflection.GeneratedProtocolMessageType('NotifyNewMemberResponse', (_message.Message,), dict(
+  DESCRIPTOR = _NOTIFYNEWMEMBERRESPONSE,
+  __module__ = 'notification_service_pb2'
   # @@protoc_insertion_point(class_scope:notification_service.NotifyNewMemberResponse)
-  })
+  ))
 _sym_db.RegisterMessage(NotifyNewMemberResponse)
 
 
-DESCRIPTOR._options = None
+DESCRIPTOR.has_options = True
+DESCRIPTOR._options = _descriptor._ParseOptions(descriptor_pb2.FileOptions(), _b('\n\035com.aiflow.notification.protoZ\025/notification_service\210\001\001\220\001\001'))
 
 _NOTIFICATIONSERVICE = _descriptor.ServiceDescriptor(
   name='NotificationService',
   full_name='notification_service.NotificationService',
   file=DESCRIPTOR,
   index=0,
-  serialized_options=None,
-  create_key=_descriptor._internal_create_key,
-  serialized_start=1729,
-  serialized_end=2864,
+  options=None,
+  serialized_start=1761,
+  serialized_end=2896,
   methods=[
   _descriptor.MethodDescriptor(
     name='sendEvent',
@@ -959,8 +955,7 @@
     containing_service=None,
     input_type=_SENDEVENTREQUEST,
     output_type=_SENDEVENTSRESPONSE,
-    serialized_options=b'\202\323\344\223\002,\"\'/aiflow/notification-service/send_event:\001*',
-    create_key=_descriptor._internal_create_key,
+    options=_descriptor._ParseOptions(descriptor_pb2.MethodOptions(), _b('\202\323\344\223\002,\"\'/aiflow/notification-service/send_event:\001*')),
   ),
   _descriptor.MethodDescriptor(
     name='listEvents',
@@ -969,8 +964,7 @@
     containing_service=None,
     input_type=_LISTEVENTSREQUEST,
     output_type=_LISTEVENTSRESPONSE,
-    serialized_options=b'\202\323\344\223\002-\"(/aiflow/notification-service/list_events:\001*',
-    create_key=_descriptor._internal_create_key,
+    options=_descriptor._ParseOptions(descriptor_pb2.MethodOptions(), _b('\202\323\344\223\002-\"(/aiflow/notification-service/list_events:\001*')),
   ),
   _descriptor.MethodDescriptor(
     name='listAllEvents',
@@ -979,8 +973,7 @@
     containing_service=None,
     input_type=_LISTALLEVENTSREQUEST,
     output_type=_LISTEVENTSRESPONSE,
-    serialized_options=b'\202\323\344\223\0021\",/aiflow/notification-service/list_all_events:\001*',
-    create_key=_descriptor._internal_create_key,
+    options=_descriptor._ParseOptions(descriptor_pb2.MethodOptions(), _b('\202\323\344\223\0021\",/aiflow/notification-service/list_all_events:\001*')),
   ),
   _descriptor.MethodDescriptor(
     name='notify',
@@ -989,8 +982,7 @@
     containing_service=None,
     input_type=_NOTIFYREQUEST,
     output_type=_NOTIFYRESPONSE,
-    serialized_options=b'\202\323\344\223\002(\"#/aiflow/notification-service/notify:\001*',
-    create_key=_descriptor._internal_create_key,
+    options=_descriptor._ParseOptions(descriptor_pb2.MethodOptions(), _b('\202\323\344\223\002(\"#/aiflow/notification-service/notify:\001*')),
   ),
   _descriptor.MethodDescriptor(
     name='listMembers',
@@ -999,8 +991,7 @@
     containing_service=None,
     input_type=_LISTMEMBERSREQUEST,
     output_type=_LISTMEMBERSRESPONSE,
-    serialized_options=b'\202\323\344\223\002.\")/aiflow/notification-service/list_members:\001*',
-    create_key=_descriptor._internal_create_key,
+    options=_descriptor._ParseOptions(descriptor_pb2.MethodOptions(), _b('\202\323\344\223\002.\")/aiflow/notification-service/list_members:\001*')),
   ),
   _descriptor.MethodDescriptor(
     name='notifyNewMember',
@@ -1009,8 +1000,7 @@
     containing_service=None,
     input_type=_NOTIFYNEWMEMBERREQUEST,
     output_type=_NOTIFYNEWMEMBERRESPONSE,
-    serialized_options=b'\202\323\344\223\0023\"./aiflow/notification-service/notify_new_member:\001*',
-    create_key=_descriptor._internal_create_key,
+    options=_descriptor._ParseOptions(descriptor_pb2.MethodOptions(), _b('\202\323\344\223\0023\"./aiflow/notification-service/notify_new_member:\001*')),
   ),
   _descriptor.MethodDescriptor(
     name='getLatestVersionByKey',
@@ -1019,8 +1009,7 @@
     containing_service=None,
     input_type=_GETLATESTVERSIONBYKEYREQUEST,
     output_type=_GETLATESTVERSIONRESPONSE,
-    serialized_options=b'\202\323\344\223\002;\"6/aiflow/notification-service/get_latest_version_by_key:\001*',
-    create_key=_descriptor._internal_create_key,
+    options=_descriptor._ParseOptions(descriptor_pb2.MethodOptions(), _b('\202\323\344\223\002;\"6/aiflow/notification-service/get_latest_version_by_key:\001*')),
   ),
 ])
 _sym_db.RegisterServiceDescriptor(_NOTIFICATIONSERVICE)
diff --git a/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2_grpc.py b/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2_grpc.py
index 1366525d7..5b0463196 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2_grpc.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/proto/notification_service_pb2_grpc.py
@@ -17,285 +17,156 @@
 # under the License.
 #
 # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
-"""Client and server classes corresponding to protobuf-defined services."""
 import grpc
 
 from . import notification_service_pb2 as notification__service__pb2
 
 
 class NotificationServiceStub(object):
-    """AirFlowService provides notification function rest endpoint of NotificationService for Notification Service component.
-    Functions of NotificationService include:
-    1.Send event.
-    2.List events.
-    """
+  """AirFlowService provides notification function rest endpoint of NotificationService for Notification Service component.
+  Functions of NotificationService include:
+  1.Send event.
+  2.List events.
+  """
 
-    def __init__(self, channel):
-        """Constructor.
+  def __init__(self, channel):
+    """Constructor.
 
-        Args:
-            channel: A grpc.Channel.
-        """
-        self.sendEvent = channel.unary_unary(
-                '/notification_service.NotificationService/sendEvent',
-                request_serializer=notification__service__pb2.SendEventRequest.SerializeToString,
-                response_deserializer=notification__service__pb2.SendEventsResponse.FromString,
-                )
-        self.listEvents = channel.unary_unary(
-                '/notification_service.NotificationService/listEvents',
-                request_serializer=notification__service__pb2.ListEventsRequest.SerializeToString,
-                response_deserializer=notification__service__pb2.ListEventsResponse.FromString,
-                )
-        self.listAllEvents = channel.unary_unary(
-                '/notification_service.NotificationService/listAllEvents',
-                request_serializer=notification__service__pb2.ListAllEventsRequest.SerializeToString,
-                response_deserializer=notification__service__pb2.ListEventsResponse.FromString,
-                )
-        self.notify = channel.unary_unary(
-                '/notification_service.NotificationService/notify',
-                request_serializer=notification__service__pb2.NotifyRequest.SerializeToString,
-                response_deserializer=notification__service__pb2.NotifyResponse.FromString,
-                )
-        self.listMembers = channel.unary_unary(
-                '/notification_service.NotificationService/listMembers',
-                request_serializer=notification__service__pb2.ListMembersRequest.SerializeToString,
-                response_deserializer=notification__service__pb2.ListMembersResponse.FromString,
-                )
-        self.notifyNewMember = channel.unary_unary(
-                '/notification_service.NotificationService/notifyNewMember',
-                request_serializer=notification__service__pb2.NotifyNewMemberRequest.SerializeToString,
-                response_deserializer=notification__service__pb2.NotifyNewMemberResponse.FromString,
-                )
-        self.getLatestVersionByKey = channel.unary_unary(
-                '/notification_service.NotificationService/getLatestVersionByKey',
-                request_serializer=notification__service__pb2.GetLatestVersionByKeyRequest.SerializeToString,
-                response_deserializer=notification__service__pb2.GetLatestVersionResponse.FromString,
-                )
+    Args:
+      channel: A grpc.Channel.
+    """
+    self.sendEvent = channel.unary_unary(
+        '/notification_service.NotificationService/sendEvent',
+        request_serializer=notification__service__pb2.SendEventRequest.SerializeToString,
+        response_deserializer=notification__service__pb2.SendEventsResponse.FromString,
+        )
+    self.listEvents = channel.unary_unary(
+        '/notification_service.NotificationService/listEvents',
+        request_serializer=notification__service__pb2.ListEventsRequest.SerializeToString,
+        response_deserializer=notification__service__pb2.ListEventsResponse.FromString,
+        )
+    self.listAllEvents = channel.unary_unary(
+        '/notification_service.NotificationService/listAllEvents',
+        request_serializer=notification__service__pb2.ListAllEventsRequest.SerializeToString,
+        response_deserializer=notification__service__pb2.ListEventsResponse.FromString,
+        )
+    self.notify = channel.unary_unary(
+        '/notification_service.NotificationService/notify',
+        request_serializer=notification__service__pb2.NotifyRequest.SerializeToString,
+        response_deserializer=notification__service__pb2.NotifyResponse.FromString,
+        )
+    self.listMembers = channel.unary_unary(
+        '/notification_service.NotificationService/listMembers',
+        request_serializer=notification__service__pb2.ListMembersRequest.SerializeToString,
+        response_deserializer=notification__service__pb2.ListMembersResponse.FromString,
+        )
+    self.notifyNewMember = channel.unary_unary(
+        '/notification_service.NotificationService/notifyNewMember',
+        request_serializer=notification__service__pb2.NotifyNewMemberRequest.SerializeToString,
+        response_deserializer=notification__service__pb2.NotifyNewMemberResponse.FromString,
+        )
+    self.getLatestVersionByKey = channel.unary_unary(
+        '/notification_service.NotificationService/getLatestVersionByKey',
+        request_serializer=notification__service__pb2.GetLatestVersionByKeyRequest.SerializeToString,
+        response_deserializer=notification__service__pb2.GetLatestVersionResponse.FromString,
+        )
 
 
 class NotificationServiceServicer(object):
-    """AirFlowService provides notification function rest endpoint of NotificationService for Notification Service component.
-    Functions of NotificationService include:
-    1.Send event.
-    2.List events.
+  """AirFlowService provides notification function rest endpoint of NotificationService for Notification Service component.
+  Functions of NotificationService include:
+  1.Send event.
+  2.List events.
+  """
+
+  def sendEvent(self, request, context):
+    """Send event.
     """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
 
-    def sendEvent(self, request, context):
-        """Send event.
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-    def listEvents(self, request, context):
-        """List events.
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-    def listAllEvents(self, request, context):
-        """List all events
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-    def notify(self, request, context):
-        """Accepts notifications from other members.
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-    def listMembers(self, request, context):
-        """List current living members.
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-    def notifyNewMember(self, request, context):
-        """Notify current members that there is a new member added.
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-    def getLatestVersionByKey(self, request, context):
-        """Get latest version by key
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-
-def add_NotificationServiceServicer_to_server(servicer, server):
-    rpc_method_handlers = {
-            'sendEvent': grpc.unary_unary_rpc_method_handler(
-                    servicer.sendEvent,
-                    request_deserializer=notification__service__pb2.SendEventRequest.FromString,
-                    response_serializer=notification__service__pb2.SendEventsResponse.SerializeToString,
-            ),
-            'listEvents': grpc.unary_unary_rpc_method_handler(
-                    servicer.listEvents,
-                    request_deserializer=notification__service__pb2.ListEventsRequest.FromString,
-                    response_serializer=notification__service__pb2.ListEventsResponse.SerializeToString,
-            ),
-            'listAllEvents': grpc.unary_unary_rpc_method_handler(
-                    servicer.listAllEvents,
-                    request_deserializer=notification__service__pb2.ListAllEventsRequest.FromString,
-                    response_serializer=notification__service__pb2.ListEventsResponse.SerializeToString,
-            ),
-            'notify': grpc.unary_unary_rpc_method_handler(
-                    servicer.notify,
-                    request_deserializer=notification__service__pb2.NotifyRequest.FromString,
-                    response_serializer=notification__service__pb2.NotifyResponse.SerializeToString,
-            ),
-            'listMembers': grpc.unary_unary_rpc_method_handler(
-                    servicer.listMembers,
-                    request_deserializer=notification__service__pb2.ListMembersRequest.FromString,
-                    response_serializer=notification__service__pb2.ListMembersResponse.SerializeToString,
-            ),
-            'notifyNewMember': grpc.unary_unary_rpc_method_handler(
-                    servicer.notifyNewMember,
-                    request_deserializer=notification__service__pb2.NotifyNewMemberRequest.FromString,
-                    response_serializer=notification__service__pb2.NotifyNewMemberResponse.SerializeToString,
-            ),
-            'getLatestVersionByKey': grpc.unary_unary_rpc_method_handler(
-                    servicer.getLatestVersionByKey,
-                    request_deserializer=notification__service__pb2.GetLatestVersionByKeyRequest.FromString,
-                    response_serializer=notification__service__pb2.GetLatestVersionResponse.SerializeToString,
-            ),
-    }
-    generic_handler = grpc.method_handlers_generic_handler(
-            'notification_service.NotificationService', rpc_method_handlers)
-    server.add_generic_rpc_handlers((generic_handler,))
-
-
- # This class is part of an EXPERIMENTAL API.
-class NotificationService(object):
-    """AirFlowService provides notification function rest endpoint of NotificationService for Notification Service component.
-    Functions of NotificationService include:
-    1.Send event.
-    2.List events.
+  def listEvents(self, request, context):
+    """List events.
     """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
 
-    @staticmethod
-    def sendEvent(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(request, target, '/notification_service.NotificationService/sendEvent',
-            notification__service__pb2.SendEventRequest.SerializeToString,
-            notification__service__pb2.SendEventsResponse.FromString,
-            options, channel_credentials,
-            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+  def listAllEvents(self, request, context):
+    """List all events
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
 
-    @staticmethod
-    def listEvents(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(request, target, '/notification_service.NotificationService/listEvents',
-            notification__service__pb2.ListEventsRequest.SerializeToString,
-            notification__service__pb2.ListEventsResponse.FromString,
-            options, channel_credentials,
-            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+  def notify(self, request, context):
+    """Accepts notifications from other members.
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
 
-    @staticmethod
-    def listAllEvents(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(request, target, '/notification_service.NotificationService/listAllEvents',
-            notification__service__pb2.ListAllEventsRequest.SerializeToString,
-            notification__service__pb2.ListEventsResponse.FromString,
-            options, channel_credentials,
-            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+  def listMembers(self, request, context):
+    """List current living members.
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
 
-    @staticmethod
-    def notify(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(request, target, '/notification_service.NotificationService/notify',
-            notification__service__pb2.NotifyRequest.SerializeToString,
-            notification__service__pb2.NotifyResponse.FromString,
-            options, channel_credentials,
-            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+  def notifyNewMember(self, request, context):
+    """Notify current members that there is a new member added.
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
 
-    @staticmethod
-    def listMembers(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(request, target, '/notification_service.NotificationService/listMembers',
-            notification__service__pb2.ListMembersRequest.SerializeToString,
-            notification__service__pb2.ListMembersResponse.FromString,
-            options, channel_credentials,
-            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+  def getLatestVersionByKey(self, request, context):
+    """Get latest version by key
+    """
+    context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+    context.set_details('Method not implemented!')
+    raise NotImplementedError('Method not implemented!')
 
-    @staticmethod
-    def notifyNewMember(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(request, target, '/notification_service.NotificationService/notifyNewMember',
-            notification__service__pb2.NotifyNewMemberRequest.SerializeToString,
-            notification__service__pb2.NotifyNewMemberResponse.FromString,
-            options, channel_credentials,
-            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
 
-    @staticmethod
-    def getLatestVersionByKey(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(request, target, '/notification_service.NotificationService/getLatestVersionByKey',
-            notification__service__pb2.GetLatestVersionByKeyRequest.SerializeToString,
-            notification__service__pb2.GetLatestVersionResponse.FromString,
-            options, channel_credentials,
-            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
+def add_NotificationServiceServicer_to_server(servicer, server):
+  rpc_method_handlers = {
+      'sendEvent': grpc.unary_unary_rpc_method_handler(
+          servicer.sendEvent,
+          request_deserializer=notification__service__pb2.SendEventRequest.FromString,
+          response_serializer=notification__service__pb2.SendEventsResponse.SerializeToString,
+      ),
+      'listEvents': grpc.unary_unary_rpc_method_handler(
+          servicer.listEvents,
+          request_deserializer=notification__service__pb2.ListEventsRequest.FromString,
+          response_serializer=notification__service__pb2.ListEventsResponse.SerializeToString,
+      ),
+      'listAllEvents': grpc.unary_unary_rpc_method_handler(
+          servicer.listAllEvents,
+          request_deserializer=notification__service__pb2.ListAllEventsRequest.FromString,
+          response_serializer=notification__service__pb2.ListEventsResponse.SerializeToString,
+      ),
+      'notify': grpc.unary_unary_rpc_method_handler(
+          servicer.notify,
+          request_deserializer=notification__service__pb2.NotifyRequest.FromString,
+          response_serializer=notification__service__pb2.NotifyResponse.SerializeToString,
+      ),
+      'listMembers': grpc.unary_unary_rpc_method_handler(
+          servicer.listMembers,
+          request_deserializer=notification__service__pb2.ListMembersRequest.FromString,
+          response_serializer=notification__service__pb2.ListMembersResponse.SerializeToString,
+      ),
+      'notifyNewMember': grpc.unary_unary_rpc_method_handler(
+          servicer.notifyNewMember,
+          request_deserializer=notification__service__pb2.NotifyNewMemberRequest.FromString,
+          response_serializer=notification__service__pb2.NotifyNewMemberResponse.SerializeToString,
+      ),
+      'getLatestVersionByKey': grpc.unary_unary_rpc_method_handler(
+          servicer.getLatestVersionByKey,
+          request_deserializer=notification__service__pb2.GetLatestVersionByKeyRequest.FromString,
+          response_serializer=notification__service__pb2.GetLatestVersionResponse.SerializeToString,
+      ),
+  }
+  generic_handler = grpc.method_handlers_generic_handler(
+      'notification_service.NotificationService', rpc_method_handlers)
+  server.add_generic_rpc_handlers((generic_handler,))
diff --git a/flink-ai-flow/lib/notification_service/notification_service/service.py b/flink-ai-flow/lib/notification_service/notification_service/service.py
index 60cd7e88b..ab924dd60 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/service.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/service.py
@@ -17,6 +17,7 @@
 # under the License.
 #
 import asyncio
+import logging
 import time
 import traceback
 
@@ -59,19 +60,20 @@ async def _send_event(self, request):
             value=event_proto.value,
             event_type=None if event_proto.event_type == "" else event_proto.event_type,
             context=None if event_proto.context == "" else event_proto.context,
-            namespace=None if event_proto.namespace == "" else event_proto.namespace)
+            namespace=None if event_proto.namespace == "" else event_proto.namespace,
+            sender=None if event_proto.sender == "" else event_proto.sender
+        )
         uuid = request.uuid
         key = event.key
-        namespace = event.namespace
         # Lock conditions dict for get/check/update of key
         await self.lock.acquire()
-        if self.notification_conditions.get((key, namespace)) is None:
-            self.notification_conditions.update({((key, namespace), asyncio.Condition())})
+        if self.notification_conditions.get(key) is None:
+            self.notification_conditions.update({(key, asyncio.Condition())})
         # Release lock after check/update key of notification conditions dict
         self.lock.release()
-        async with self.notification_conditions.get((key, namespace)), self.write_condition:
+        async with self.notification_conditions.get(key), self.write_condition:
             event: BaseEvent = self.storage.add_event(event, uuid)
-            self.notification_conditions.get((key, namespace)).notify_all()
+            self.notification_conditions.get(key).notify_all()
             self.write_condition.notify_all()
 
         result_event_proto = event_to_proto(event)
@@ -93,11 +95,12 @@ async def _list_events(self, request):
         event_type = request.event_type
         start_time = request.start_time
         start_version = request.start_version
-        namespace = request.namespace
+        namespace = None if request.namespace == '' else request.namespace
+        sender = None if request.sender == '' else request.sender
         timeout_seconds = request.timeout_seconds
 
         if timeout_seconds == 0:
-            event_models = self._query_events(keys, event_type, start_time, start_version, namespace)
+            event_models = self._query_events(keys, event_type, start_time, start_version, namespace, sender)
             event_proto_list = event_list_to_proto(event_models)
             return notification_service_pb2.ListEventsResponse(
                 return_code=notification_service_pb2.ReturnStatus.SUCCESS,
@@ -108,14 +111,14 @@ async def _list_events(self, request):
             # Lock conditions dict for get/check/update of key
             await self.lock.acquire()
             for key in keys:
-                if self.notification_conditions.get((key, namespace)) is None:
-                    self.notification_conditions.update({((key, namespace), asyncio.Condition())})
+                if self.notification_conditions.get(key) is None:
+                    self.notification_conditions.update({(key, asyncio.Condition())})
             # Release lock after check/update key of notification conditions dict
             self.lock.release()
             event_models = []
             if len(keys) == 1:
                 key = keys[0]
-                condition = self.notification_conditions.get((key, namespace))
+                condition = self.notification_conditions.get(key)
             else:
                 condition = self.write_condition
             async with condition:
@@ -124,20 +127,20 @@ async def _list_events(self, request):
                         await asyncio.wait_for(condition.wait(),
                                                timeout_seconds - time.time() + start)
                         event_models = self._query_events(
-                            keys, event_type, start_time, start_version, namespace)
+                            keys, event_type, start_time, start_version, namespace, sender)
                     except asyncio.TimeoutError:
                         pass
                 if len(event_models) == 0:
                     event_models = self._query_events(
-                        keys, event_type, start_time, start_version, namespace)
+                        keys, event_type, start_time, start_version, namespace, sender)
             event_proto_list = event_list_to_proto(event_models)
             return notification_service_pb2.ListEventsResponse(
                 return_code=notification_service_pb2.ReturnStatus.SUCCESS,
                 return_msg='',
                 events=event_proto_list)
 
-    def _query_events(self, keys, event_type, start_time, start_version, namespace):
-        return self.storage.list_events(keys, start_version, event_type, start_time, namespace)
+    def _query_events(self, keys, event_type, start_time, start_version, namespace, sender):
+        return self.storage.list_events(keys, start_version, event_type, start_time, namespace, sender)
 
     @asyncio.coroutine
     def listAllEvents(self, request, context):
@@ -231,9 +234,9 @@ def notify(self, request, context):
 
     async def _notify(self, request):
         for notify in request.notifies:
-            if (notify.key, notify.namespace) in self.notification_conditions:
-                async with self.notification_conditions.get((notify.key, notify.namespace)):
-                    self.notification_conditions.get((notify.key, notify.namespace)).notify_all()
+            if notify.key in self.notification_conditions:
+                async with self.notification_conditions.get(notify.key):
+                    self.notification_conditions.get(notify.key).notify_all()
         async with self.write_condition:
             self.write_condition.notify_all()
         return notification_service_pb2.NotifyResponse(
diff --git a/flink-ai-flow/lib/notification_service/notification_service/util/db.py b/flink-ai-flow/lib/notification_service/notification_service/util/db.py
index eee48adca..c88a13504 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/util/db.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/util/db.py
@@ -108,6 +108,7 @@ class EventModel(Base):
     event_type = Column(String(1024), server_default="UNDEFINED")
     context = Column(Text())
     namespace = Column(String(1024))
+    sender = Column(String(1024))
     create_time = Column(BigInteger(), nullable=False)
     uuid = Column(String(40), nullable=False, unique=True)
 
@@ -122,6 +123,7 @@ def add_event(event: BaseEvent, uuid, session=None):
         event_model.namespace = event.namespace
         event_model.create_time = int(time.time() * 1000)
         event_model.uuid = uuid
+        event_model.sender = event.sender
         session.add(event_model)
         session.commit()
         return event_model_to_event(event_model)
@@ -133,10 +135,12 @@ def list_events(key: Union[str, Tuple[str]],
                     event_type: str = None,
                     start_time: int = None,
                     namespace: str = None,
+                    sender: str = None,
                     session=None):
         key = None if key == "" else key
         event_type = None if event_type == "" else event_type
         namespace = None if namespace == "" else namespace
+        sender = None if sender == "" else sender
         if isinstance(key, str):
             key = (key,)
         elif isinstance(key, Iterable):
@@ -149,8 +153,10 @@ def list_events(key: Union[str, Tuple[str]],
             conditions.append(EventModel.event_type == event_type)
         if start_time is not None and start_time > 0:
             conditions.append(EventModel.create_time >= start_time)
-        if ANY_CONDITION != namespace:
+        if namespace is not None and ANY_CONDITION != namespace:
             conditions.append(EventModel.namespace == namespace)
+        if sender is not None and ANY_CONDITION != sender:
+            conditions.append(EventModel.sender == sender)
         if version > 0:
             conditions.append(EventModel.version > version)
         if ANY_CONDITION not in key:
@@ -189,6 +195,7 @@ def sync_event(event: BaseEvent, uuid, session=None):
         event_model.namespace = event.namespace
         event_model.create_time = event.create_time
         event_model.uuid = uuid
+        event_model.sender = event.sender
         session.add(event_model)
         session.commit()
         return event_model_to_event(event_model)
diff --git a/flink-ai-flow/lib/notification_service/notification_service/util/utils.py b/flink-ai-flow/lib/notification_service/notification_service/util/utils.py
index 537e738ba..16ef56c63 100644
--- a/flink-ai-flow/lib/notification_service/notification_service/util/utils.py
+++ b/flink-ai-flow/lib/notification_service/notification_service/util/utils.py
@@ -32,7 +32,8 @@ def event_to_proto(event: BaseEvent):
                                                              event_type=event.event_type,
                                                              create_time=event.create_time,
                                                              namespace=event.namespace,
-                                                             context=event.context)
+                                                             context=event.context,
+                                                             sender=event.sender)
     return result_event_proto
 
 
@@ -51,7 +52,8 @@ def event_proto_to_event(event_proto):
                      version=event_proto.version,
                      create_time=event_proto.create_time,
                      context=event_proto.context,
-                     namespace=event_proto.namespace)
+                     namespace=event_proto.namespace,
+                     sender=event_proto.sender)
 
 
 def event_model_to_event(event_model):
@@ -62,7 +64,9 @@ def event_model_to_event(event_model):
         version=event_model.version,
         create_time=event_model.create_time,
         context=event_model.context,
-        namespace=event_model.namespace)
+        namespace=event_model.namespace,
+        sender=event_model.sender
+    )
 
 
 def member_to_proto(member: Member):


Test patch:
diff --git a/flink-ai-flow/ai_flow/test/airflow/test_dag_generator.py b/flink-ai-flow/ai_flow/test/airflow/test_dag_generator.py
index 57ff37eed..1b17a8e90 100644
--- a/flink-ai-flow/ai_flow/test/airflow/test_dag_generator.py
+++ b/flink-ai-flow/ai_flow/test/airflow/test_dag_generator.py
@@ -18,21 +18,49 @@
 #
 import unittest
 
-from ai_flow.workflow.job_config import BaseJobConfig, PeriodicConfig
-
 from ai_flow.airflow.dag_generator import DAGGenerator
-from ai_flow.project.project_description import ProjectDesc
-from ai_flow.meta.job_meta import State
 from ai_flow.graph.edge import JobControlEdge, MetConfig, generate_job_status_key
-from ai_flow.plugins.local_dummy_job_plugin import LocalDummyJob, SendEventJobConfig
+from ai_flow.meta.job_meta import State
 from ai_flow.plugins.local_cmd_job_plugin import LocalCMDJob
+from ai_flow.plugins.local_dummy_job_plugin import LocalDummyJob, SendEventJobConfig
+from ai_flow.project.project_description import ProjectDesc
 from ai_flow.workflow.job import BaseJob
-from notification_service.base_notification import UNDEFINED_EVENT_TYPE
+from ai_flow.workflow.job_config import BaseJobConfig, PeriodicConfig
 from ai_flow.workflow.workflow import Workflow
+from notification_service.base_notification import UNDEFINED_EVENT_TYPE
 
 
 class TestDAGGenerator(unittest.TestCase):
 
+    def test_generate_bash_dag_code(self):
+        from datetime import datetime, timedelta
+        default_args = {
+            'owner': 'airflow',
+            'depends_on_past': False,
+            'start_date': datetime(2015, 12, 1),
+            'email': ['airflow@example.com'],
+            'email_on_failure': False,
+            'email_on_retry': False,
+            'retries': 1,
+            'retry_delay': timedelta(minutes=5),
+            'schedule_interval': None,
+        }
+        generator = DAGGenerator()
+        workflow = TestDAGGenerator.create_bash_workflow()
+        dag = generator.generator(workflow, 'aa', default_args)
+        self.assertIsNotNone(dag)
+        for i in range(3):
+            self.assertTrue(
+                "BashOperator(task_id='{0}_job', dag=dag, bash_command='echo \"{0} hello word!\"')".format(i) in dag)
+
+    def test_generate_dummy_dag_code(self):
+        generator = DAGGenerator()
+        workflow = TestDAGGenerator.create_workflow()
+        dag = generator.generator(workflow)
+        self.assertIsNotNone(dag)
+        self.assertFalse('DummyOperator' in dag)
+        self.assertFalse('SendEventOperator' in dag)
+
     @staticmethod
     def create_dummy_job(index) -> BaseJob:
         job: BaseJob = LocalDummyJob()
@@ -56,41 +84,31 @@ def create_workflow() -> Workflow:
             elif i == 5:
                 job.job_config = SendEventJobConfig('localhost:50051', 'key_2', 'value_2', "STOP_SCHEDULER_CMD")
             workflow.add_job(job)
-        deps = []
-        deps.append(JobControlEdge(target_node_id='0_job', source_node_id='2_job',
-                                   met_config=MetConfig(event_key=generate_job_status_key('0_job'),
-                                                        event_value=State.FINISHED.value)))
-        deps.append(JobControlEdge(target_node_id='1_job', source_node_id='2_job',
-                                   met_config=MetConfig(event_key=generate_job_status_key('1_job'),
-                                                        event_value=State.FINISHED.value)))
-        workflow.add_edges("2_job", deps)
+        dependencies = [JobControlEdge(target_node_id='0_job', source_node_id='2_job',
+                                       met_config=MetConfig(event_key=generate_job_status_key('0_job'),
+                                                            event_value=State.FINISHED.value)),
+                        JobControlEdge(target_node_id='1_job', source_node_id='2_job',
+                                       met_config=MetConfig(event_key=generate_job_status_key('1_job'),
+                                                            event_value=State.FINISHED.value))]
+        workflow.add_edges("2_job", dependencies)
 
-        deps = []
-        deps.append(JobControlEdge(target_node_id='2_job', source_node_id='4_job',
-                                   met_config=MetConfig(event_key='key_1',
-                                                        event_value='value_1',
-                                                        event_type=UNDEFINED_EVENT_TYPE)))
-        deps.append(JobControlEdge(target_node_id='3_job', source_node_id='4_job',
-                                   met_config=MetConfig(event_key='key_2',
-                                                        event_value='value_2',
-                                                        event_type=UNDEFINED_EVENT_TYPE)))
-        workflow.add_edges("4_job", deps)
+        dependencies = [JobControlEdge(target_node_id='2_job', source_node_id='4_job',
+                                       met_config=MetConfig(event_key='key_1',
+                                                            event_value='value_1',
+                                                            event_type=UNDEFINED_EVENT_TYPE)),
+                        JobControlEdge(target_node_id='3_job', source_node_id='4_job',
+                                       met_config=MetConfig(event_key='key_2',
+                                                            event_value='value_2',
+                                                            event_type=UNDEFINED_EVENT_TYPE))]
+        workflow.add_edges("4_job", dependencies)
 
-        deps = []
-        deps.append(JobControlEdge(target_node_id='4_job', source_node_id='5_job',
-                                   met_config=MetConfig(event_key=generate_job_status_key('5_job'),
-                                                        event_value=State.FINISHED.value)))
-        workflow.add_edges("5_job", deps)
+        dependencies = [JobControlEdge(target_node_id='4_job', source_node_id='5_job',
+                                       met_config=MetConfig(event_key=generate_job_status_key('5_job'),
+                                                            event_value=State.FINISHED.value))]
+        workflow.add_edges("5_job", dependencies)
         workflow.workflow_id = 1
         return workflow
 
-    def test_generate_dummy_dag_code(self):
-        generator = DAGGenerator()
-        workflow = TestDAGGenerator.create_workflow()
-        dag = generator.generator(workflow)
-        self.assertFalse('DummyOperator' in dag)
-        self.assertFalse('SendEventOperator' in dag)
-
     @staticmethod
     def create_bash_job(index) -> BaseJob:
         job: BaseJob = LocalCMDJob(exec_cmd=['echo "{0} hello word!"'.format(index)])
@@ -115,35 +133,16 @@ def create_bash_workflow() -> Workflow:
         for i in range(3):
             job = TestDAGGenerator.create_bash_job(i)
             workflow.add_job(job)
-        deps_1 = [JobControlEdge(target_node_id='', source_node_id='1_job',
-                                 met_config=MetConfig(event_key='key_1',
-                                                      event_value='value_1',
-                                                      event_type=UNDEFINED_EVENT_TYPE))]
-        deps_2 = [JobControlEdge(target_node_id='', source_node_id='2_job',
-                                 met_config=MetConfig(event_key='key_2',
-                                                      event_value='value_2',
-                                                      event_type=UNDEFINED_EVENT_TYPE))]
+        dependencies_1 = [JobControlEdge(target_node_id='', source_node_id='1_job',
+                                         met_config=MetConfig(event_key='key_1',
+                                                              event_value='value_1',
+                                                              event_type=UNDEFINED_EVENT_TYPE))]
+        dependencies_2 = [JobControlEdge(target_node_id='1_job', source_node_id='2_job',
+                                         met_config=MetConfig(event_key='key_2',
+                                                              event_value='value_2',
+                                                              event_type=UNDEFINED_EVENT_TYPE))]
 
-        workflow.add_edges("1_job", deps_1)
-        workflow.add_edges("2_job", deps_2)
+        workflow.add_edges("1_job", dependencies_1)
+        workflow.add_edges("2_job", dependencies_2)
         workflow.workflow_id = 1
         return workflow
-
-    def test_generate_bash_dag_code(self):
-        from datetime import datetime, timedelta
-        default_args = {
-            'owner': 'airflow',
-            'depends_on_past': False,
-            'start_date': datetime(2015, 12, 1),
-            'email': ['airflow@example.com'],
-            'email_on_failure': False,
-            'email_on_retry': False,
-            'retries': 1,
-            'retry_delay': timedelta(minutes=5),
-            'schedule_interval': None,
-        }
-        generator = DAGGenerator()
-        workflow = TestDAGGenerator.create_bash_workflow()
-        dag = generator.generator(workflow, 'aa', default_args)
-        print("\n\n")
-        print(dag)
diff --git a/flink-ai-flow/lib/airflow/tests/contrib/jobs/test_event_based_scheduler.py b/flink-ai-flow/lib/airflow/tests/contrib/jobs/test_event_based_scheduler.py
index c4de90719..5f6d34a30 100644
--- a/flink-ai-flow/lib/airflow/tests/contrib/jobs/test_event_based_scheduler.py
+++ b/flink-ai-flow/lib/airflow/tests/contrib/jobs/test_event_based_scheduler.py
@@ -301,7 +301,7 @@ def run_trigger_task_function(self):
                 tes = session.query(TaskExecution).filter(TaskExecution.dag_id == 'trigger_task',
                                                           TaskExecution.task_id == 'task_1').all()
                 if len(tes) > 0:
-                    client.schedule_task('task_2', SchedulingAction.START, execution_context)
+                    client.schedule_task('trigger_task', 'task_2', SchedulingAction.START, execution_context)
                     while True:
                         with create_session() as session_2:
                             tes_2 = session_2.query(TaskExecution).filter(TaskExecution.dag_id == 'trigger_task',
diff --git a/flink-ai-flow/lib/airflow/tests/dags/test_event_based_scheduler.py b/flink-ai-flow/lib/airflow/tests/dags/test_event_based_scheduler.py
index 0e89ecacf..4435c4e9c 100644
--- a/flink-ai-flow/lib/airflow/tests/dags/test_event_based_scheduler.py
+++ b/flink-ai-flow/lib/airflow/tests/dags/test_event_based_scheduler.py
@@ -41,7 +41,7 @@
     dag=dag,
     event_handler=ActionEventHandler()
 )
-t2.subscribe_event('start', '', event_namespace='test_namespace')
+t2.subscribe_event('start', '', event_namespace='test_namespace', from_task_id='print_date')
 t2.subscribe_event(event_key="stop",
                    event_namespace='test_namespace',
                    event_type=UNDEFINED_EVENT_TYPE)
@@ -70,7 +70,8 @@ def my_sleeping_function(random_base):
 t3.subscribe_event('start', '', event_namespace='test_namespace')
 t3.subscribe_event(event_key="restart",
                    event_namespace='test_namespace',
-                   event_type=UNDEFINED_EVENT_TYPE)
+                   event_type=UNDEFINED_EVENT_TYPE,
+                   from_task_id='sleep_to_be_stopped')
 
 dag.doc_md = __doc__
 
diff --git a/flink-ai-flow/lib/airflow/tests/dags/test_task_trigger_dag.py b/flink-ai-flow/lib/airflow/tests/dags/test_task_trigger_dag.py
index 26620025f..b6e9842a1 100644
--- a/flink-ai-flow/lib/airflow/tests/dags/test_task_trigger_dag.py
+++ b/flink-ai-flow/lib/airflow/tests/dags/test_task_trigger_dag.py
@@ -24,5 +24,5 @@
 op1 = BashOperator(task_id="task_1", dag=dag,  owner='airflow', bash_command='echo "hello world 1!"')
 
 op2 = BashOperator(task_id="task_2", dag=dag,  owner='airflow', bash_command='echo "hello world 2!"')
-op2.subscribe_event('key_1')
+op2.subscribe_event(event_key='key_1', from_task_id='task_1')
 op2.set_events_handler(StartEventHandler())
diff --git a/flink-ai-flow/lib/airflow/tests/models/test_dag.py b/flink-ai-flow/lib/airflow/tests/models/test_dag.py
index 89a8645ec..207b63bde 100644
--- a/flink-ai-flow/lib/airflow/tests/models/test_dag.py
+++ b/flink-ai-flow/lib/airflow/tests/models/test_dag.py
@@ -77,10 +77,13 @@ def test_create_dependencies(self):
     def test_dependencies_json(self):
         dag_event_dependencies = DagEventDependencies()
         dag_event_dependencies.add_dependencies('task_1', EventKey(key='key_1', event_type='type_1'))
-        dag_event_dependencies.add_dependencies('task_2', EventKey(key='key_1', event_type='type_2'))
+        dag_event_dependencies.add_dependencies('task_2', EventKey(key='key_1', event_type='type_2'),
+                                                from_task_id='task_1')
         dag_event_dependencies.add_dependencies('task_3', EventKey(key='key_1', event_type='*'))
         json_str = DagEventDependencies.to_json(dag_event_dependencies)
         dag_event_dependencies_copy: DagEventDependencies = DagEventDependencies.from_json(json_str)
+        self.assertEqual('task_1',
+                         dag_event_dependencies_copy.task_event_dependencies['default']['key_1']['type_2']['task_2'])
         dag_event_dependencies_copy.add_dependencies('task_4', EventKey(key='key_1', event_type='type_1'))
         tasks = dag_event_dependencies_copy.find_affected_tasks(EventKey('key_1', 'type_1'))
         self.assertEqual(3, len(tasks))
diff --git a/flink-ai-flow/lib/notification_service/tests/test_notification.py b/flink-ai-flow/lib/notification_service/tests/test_notification.py
index 564ac620b..64b0319be 100644
--- a/flink-ai-flow/lib/notification_service/tests/test_notification.py
+++ b/flink-ai-flow/lib/notification_service/tests/test_notification.py
@@ -50,6 +50,7 @@ def test_send_event(self):
 
     def test_list_events(self):
         self.client._default_namespace = "a"
+        self.client._sender = 's'
         event1 = self.client.send_event(BaseEvent(key="key", value="value1"))
 
         self.client._default_namespace = "b"
@@ -59,6 +60,7 @@ def test_list_events(self):
 
         events = self.client.list_events(["key", "key2"], version=event1.version)
         self.assertEqual(3, len(events))
+        self.assertEqual('s', events[0].sender)
 
         self.client._default_namespace = "a"
         events = self.client.list_events("key")
@@ -71,6 +73,12 @@ def test_list_events(self):
         events = self.client.list_events("key", event_type="a")
         self.assertEqual(1, len(events))
 
+        events = self.client.list_events("key", sender='s')
+        self.assertEqual(2, len(events))
+
+        events = self.client.list_events("key", sender='p')
+        self.assertEqual(0, len(events))
+
     def test_listen_events(self):
         event_list = []
 
@@ -83,10 +91,11 @@ def process(self, events: List[BaseEvent]):
                 self.event_list.extend(events)
 
         self.client._default_namespace = "a"
+        self.client._sender = "s"
         event1 = self.client.send_event(BaseEvent(key="key", value="value1"))
-        self.client.start_listen_event(key="key",
-                                       watcher=TestWatch(event_list),
-                                       version=event1.version)
+        h = self.client.start_listen_event(key="key",
+                                           watcher=TestWatch(event_list),
+                                           version=event1.version)
         self.client.send_event(BaseEvent(key="key", value="value2"))
         self.client.send_event(BaseEvent(key="key", value="value3"))
 
@@ -94,11 +103,85 @@ def process(self, events: List[BaseEvent]):
         self.client.send_event(BaseEvent(key="key", value="value4"))
 
         self.client._default_namespace = "a"
-        self.client.stop_listen_event("key")
+        h.stop()
         events = self.client.list_events("key", version=event1.version)
         self.assertEqual(2, len(events))
         self.assertEqual(2, len(event_list))
 
+        # listen by event_type
+        print('listen by event_type')
+        event_list.clear()
+        h = self.client.start_listen_event(key="key",
+                                           watcher=TestWatch(event_list),
+                                           start_time=int(time.time() * 1000),
+                                           event_type='e')
+        self.client.send_event(BaseEvent(key="key", value="value2", event_type='e'))
+        self.client.send_event(BaseEvent(key="key", value="value2", event_type='f'))
+        h.stop()
+        self.assertEqual(1, len(event_list))
+
+        event_list.clear()
+        h = self.client.start_listen_event(key="key",
+                                           start_time=int(time.time() * 1000),
+                                           watcher=TestWatch(event_list))
+        self.client.send_event(BaseEvent(key="key", value="value2", event_type='e'))
+        self.client.send_event(BaseEvent(key="key", value="value2", event_type='f'))
+        h.stop()
+        self.assertEqual(2, len(event_list))
+
+        # listen by namespace
+        print("listen by namespace")
+        self.client._default_namespace = "a"
+        event_list.clear()
+        h = self.client.start_listen_event(key="key",
+                                           start_time=int(time.time() * 1000),
+                                           watcher=TestWatch(event_list),
+                                           namespace='a')
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        self.client._default_namespace = "b"
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        h.stop()
+        self.assertEqual(1, len(event_list))
+
+        event_list.clear()
+        h = self.client.start_listen_event(key="key",
+                                           start_time=int(time.time() * 1000),
+                                           watcher=TestWatch(event_list),
+                                           namespace='*')
+        self.client._default_namespace = "a"
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        self.client._default_namespace = "b"
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        h.stop()
+        self.assertEqual(2, len(event_list))
+
+        # listen by sender
+        print("listen by sender")
+        event_list.clear()
+        h = self.client.start_listen_event(key="key",
+                                           watcher=TestWatch(event_list),
+                                           start_time=int(time.time() * 1000),
+                                           namespace='*',
+                                           sender='s')
+        self.client._sender = "s"
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        self.client._sender = "p"
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        h.stop()
+        self.assertEqual(1, len(event_list))
+
+        event_list.clear()
+        h = self.client.start_listen_event(key="key",
+                                           watcher=TestWatch(event_list),
+                                           start_time=int(time.time() * 1000),
+                                           namespace='*')
+        self.client._sender = "s"
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        self.client._sender = "p"
+        self.client.send_event(BaseEvent(key="key", value="value2"))
+        h.stop()
+        self.assertEqual(2, len(event_list))
+
     def test_all_listen_events(self):
         self.client.send_event(BaseEvent(key="key", value="value1"))
         event2 = self.client.send_event(BaseEvent(key="key", value="value2"))
@@ -164,11 +247,13 @@ def test_get_latest_version(self):
 
     def test_list_any_condition(self):
         self.client._default_namespace = 'a'
+        self.client._sender = 's'
         self.client.send_event(BaseEvent(key="key_1", value="value1"))
         self.client.send_event(BaseEvent(key="key_2", value="value2"))
         result = self.client.list_events(key='*', event_type='*')
         self.assertEqual(2, len(result))
         self.client._default_namespace = 'b'
+        self.client._sender = 'p'
         self.client.send_event(BaseEvent(key="key_1", value="value1", event_type='event_type'))
         result = self.client.list_events(key='*', event_type='*')
         self.assertEqual(1, len(result))
@@ -178,7 +263,15 @@ def test_list_any_condition(self):
         self.assertEqual(2, len(result))
         result = self.client.list_events(key='key_1', event_type='event_type', namespace='*')
         self.assertEqual(1, len(result))
-        
+        result = self.client.list_events(key='key_1', namespace='*')
+        self.assertEqual(2, len(result))
+        result = self.client.list_events(key='key_1', namespace='*', sender='*')
+        self.assertEqual(2, len(result))
+        result = self.client.list_events(key='key_1', namespace='*', sender='s')
+        self.assertEqual(1, len(result))
+        result = self.client.list_events(key='key_1', namespace='*', sender='p')
+        self.assertEqual(1, len(result))
+
 
 class DbStorageTest(unittest.TestCase, NotificationTest):