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

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


Discussion / hints:

Fix patch:
diff --git a/flink-ai-flow/lib/notification_service/java/pom.xml b/flink-ai-flow/lib/notification_service/java/pom.xml
index 6bb03958d..e19166aaf 100644
--- a/flink-ai-flow/lib/notification_service/java/pom.xml
+++ b/flink-ai-flow/lib/notification_service/java/pom.xml
@@ -32,6 +32,7 @@
         <protoc.version>3.11.0</protoc.version>
         <commons.lang3.version>3.9</commons.lang3.version>
         <commons.collections4.version>4.4</commons.collections4.version>
+        <slf4j.version>1.7.30</slf4j.version>
         <junit.version>4.12</junit.version>
         <mockito.version>2.28.2</mockito.version>
         <jython.version>2.7.0</jython.version>
@@ -113,6 +114,11 @@
             <artifactId>jython</artifactId>
             <version>${jython.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java
index 19608d691..1cf875065 100644
--- a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/EventListener.java
@@ -22,7 +22,6 @@
 import com.aiflow.notification.proto.NotificationServiceGrpc.NotificationServiceBlockingStub;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.commons.collections4.CollectionUtils;
-import sun.util.logging.resources.logging;
 
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -94,7 +93,6 @@ public Runnable listenEvents() {
 									this.startTime,
 									this.namespace,
 									this.timeoutSeconds);
-					System.out.println("######" + events.size());
 					if (CollectionUtils.isNotEmpty(events)) {
 						this.watcher.process(events);
 						listenVersion = events.get(events.size() - 1).getVersion();
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java
index 86a6c1c6d..fb87536fc 100644
--- a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationClient.java
@@ -20,243 +20,371 @@
 
 import com.aiflow.notification.entity.EventMeta;
 import com.aiflow.notification.proto.NotificationServiceGrpc.NotificationServiceBlockingStub;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.EventProto;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.GetLatestVersionByKeyRequest;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.GetLatestVersionResponse;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.ListAllEventsRequest;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.ListEventsRequest;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.ListEventsResponse;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.ReturnStatus;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.SendEventRequest;
-import com.aiflow.notification.proto.NotificationServiceOuterClass.SendEventsResponse;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.*;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.grpc.ManagedChannelBuilder;
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import static com.aiflow.notification.entity.EventMeta.buildEventMeta;
 import static com.aiflow.notification.proto.NotificationServiceGrpc.newBlockingStub;
 
 public class NotificationClient {
 
-	public static final String SERVER_URI = "localhost:50051";
+    private static final Logger logger = LoggerFactory.getLogger(NotificationClient.class);
+    private static final String SERVER_URI = "localhost:50051";
+    private final String defaultNamespace;
+    private final Integer listMemberIntervalMs;
+    private final Integer retryIntervalMs;
+    private final Integer retryTimeoutMs;
+    private final Map<Map<String, String>, EventListener> threads;
+    private final ExecutorService listMembersService;
+    private NotificationServiceBlockingStub notificationServiceStub;
+    private Set<MemberProto> livingMembers;
+    private Boolean enableHa;
+    private String currentUri;
 
-	private final String defaultNamespace;
-	private final Boolean enableHa;
-	private final Integer listMemberIntervalMs;
-	private final Integer retryIntervalMs;
-	private final Integer retryTimeoutMs;
-	private final NotificationServiceBlockingStub notificationServiceStub;
-	private final Map<Map<String, String>, EventListener> threads;
+    public NotificationClient(
+            String target,
+            String defaultNamespace,
+            Boolean enableHa,
+            Integer listMemberIntervalMs,
+            Integer retryIntervalMs,
+            Integer retryTimeoutMs) {
+        this.defaultNamespace = defaultNamespace;
+        this.enableHa = enableHa;
+        this.listMemberIntervalMs = listMemberIntervalMs;
+        this.retryIntervalMs = retryIntervalMs;
+        this.retryTimeoutMs = retryTimeoutMs;
+        if (enableHa) {
+            String[] serverUris = StringUtils.split(target, ",");
+            boolean lastError = true;
+            for (String serverUri : serverUris) {
+                currentUri = serverUri;
+                try {
+                    initNotificationServiceStub();
+                    lastError = false;
+                    break;
+                } catch (Exception e) {
+                    continue;
+                }
+            }
+            if (lastError) {
+                logger.warn("Failed to initialize client");
+            }
+        } else {
+            currentUri = target;
+            initNotificationServiceStub();
+        }
+        threads = new HashMap<>();
+        livingMembers = new HashSet<>();
+        listMembersService =
+                Executors.newSingleThreadExecutor(
+                        new ThreadFactoryBuilder()
+                                .setDaemon(true)
+                                .setNameFormat("list-members-%d")
+                                .build());
+        listMembersService.submit(listMembers());
+    }
 
-	public NotificationClient(
-			String target,
-			String defaultNamespace,
-			Boolean enableHa,
-			Integer listMemberIntervalMs,
-			Integer retryIntervalMs,
-			Integer retryTimeoutMs) {
-		this.defaultNamespace = defaultNamespace;
-		this.enableHa = enableHa;
-		this.listMemberIntervalMs = listMemberIntervalMs;
-		this.retryIntervalMs = retryIntervalMs;
-		this.retryTimeoutMs = retryTimeoutMs;
-		this.notificationServiceStub =
-				newBlockingStub(
-						ManagedChannelBuilder.forTarget(
-								StringUtils.isEmpty(target) ? SERVER_URI : target)
-								.usePlaintext()
-								.build());
-		this.threads = new HashMap<>();
-	}
+    /**
+     * List specific registered listener events in Notification Service.
+     *
+     * @param serviceStub Notification service GRPC stub.
+     * @param keys Keys of event for listening.
+     * @param version (Optional) Version of event for listening.
+     * @param timeoutSeconds List events request timeout seconds.
+     * @return List of event updated in Notification Service.
+     */
+    protected static List<EventMeta> listEvents(
+            NotificationServiceBlockingStub serviceStub,
+            List<String> keys,
+            long version,
+            String eventType,
+            long startTime,
+            String namespace,
+            Integer timeoutSeconds)
+            throws Exception {
+        ListEventsRequest request =
+                ListEventsRequest.newBuilder()
+                        .addAllKeys(keys)
+                        .setStartVersion(version)
+                        .setEventType(eventType)
+                        .setStartTime(startTime)
+                        .setNamespace(namespace)
+                        .setTimeoutSeconds(timeoutSeconds)
+                        .build();
+        return parseEventsFromResponse(serviceStub.listEvents(request));
+    }
 
-	/**
-	 * Send the event to Notification Service.
-	 *
-	 * @param key Key of event updated in Notification Service.
-	 * @param value Value of event updated in Notification Service.
-	 * @param eventType Type of event updated in Notification Service.
-	 * @param context Context of event updated in Notification Service.
-	 * @return Object of Event created in Notification Service.
-	 */
-	public EventMeta sendEvent(String key, String value, String eventType, String context)
-			throws Exception {
-		SendEventRequest request =
-				SendEventRequest.newBuilder()
-						.setEvent(
-								EventProto.newBuilder()
-										.setKey(key)
-										.setValue(value)
-										.setEventType(eventType)
-										.setContext(context)
-										.setNamespace(defaultNamespace)
-										.build())
-						.setUuid(UUID.randomUUID().toString())
-						.build();
-		SendEventsResponse response = this.notificationServiceStub.sendEvent(request);
-		if (response.getReturnCode() == ReturnStatus.SUCCESS) {
-			return buildEventMeta(response.getEvent());
-		} else {
-			throw new Exception(response.getReturnMsg());
-		}
-	}
+    private static List<EventMeta> parseEventsFromResponse(ListEventsResponse response)
+            throws Exception {
+        if (response.getReturnCode() == ReturnStatus.SUCCESS) {
+            List<EventMeta> eventMetas = new ArrayList<>();
+            for (EventProto eventProto : response.getEventsList()) {
+                eventMetas.add(buildEventMeta(eventProto));
+            }
+            return eventMetas;
+        } else {
+            throw new Exception(response.getReturnMsg());
+        }
+    }
 
-	/**
-	 * List specific `key` or `version` notifications in Notification Service.
-	 *
-	 * @param keys Keys of notification for listening.
-	 * @param version (Optional) Version of notification for listening.
-	 * @param eventType (Optional) Type of event for listening.
-	 * @param startTime (Optional) Type of event for listening.
-	 * @return List of Notification updated in Notification Service.
-	 */
-	public List<EventMeta> listEvents(
-			List<String> keys, long version, String eventType, long startTime) throws Exception {
-		return listEvents(
-				this.notificationServiceStub,
-				keys,
-				version,
-				eventType,
-				startTime,
-				defaultNamespace,
-				0);
-	}
+    protected static NotificationServiceBlockingStub wrapBlockingStub(
+            NotificationServiceBlockingStub stub,
+            String target,
+            Set<MemberProto> livingMembers,
+            Boolean haRunning,
+            Integer retryIntervalMs,
+            Integer retryTimeoutMs) {
+        return newBlockingStub(ManagedChannelBuilder.forTarget(target).usePlaintext().build())
+                .withInterceptors(
+                        new NotificationInterceptor(
+                                stub,
+                                target,
+                                livingMembers,
+                                haRunning,
+                                retryIntervalMs,
+                                retryTimeoutMs));
+    }
 
-	/**
-	 * List specific registered listener events in Notification Service.
-	 *
-	 * @param serviceStub Notification service GRPC stub.
-	 * @param keys Keys of event for listening.
-	 * @param version (Optional) Version of event for listening.
-	 * @param timeoutSeconds List events request timeout seconds.
-	 * @return List of event updated in Notification Service.
-	 */
-	protected static List<EventMeta> listEvents(
-			NotificationServiceBlockingStub serviceStub,
-			List<String> keys,
-			long version,
-			String eventType,
-			long startTime,
-			String namespace,
-			Integer timeoutSeconds)
-			throws Exception {
-		ListEventsRequest request =
-				ListEventsRequest.newBuilder()
-						.addAllKeys(keys)
-						.setStartVersion(version)
-						.setEventType(eventType)
-						.setStartTime(startTime)
-						.setNamespace(namespace)
-						.setTimeoutSeconds(timeoutSeconds)
-						.build();
-		return parseEventsFromResponse(serviceStub.listEvents(request));
-	}
+    /** Initialize notification service stub. */
+    protected void initNotificationServiceStub() {
+        notificationServiceStub =
+                newBlockingStub(
+                        ManagedChannelBuilder.forTarget(
+                                        StringUtils.isEmpty(currentUri) ? SERVER_URI : currentUri)
+                                .usePlaintext()
+                                .build());
+        notificationServiceStub =
+                wrapBlockingStub(
+                        notificationServiceStub,
+                        StringUtils.isEmpty(currentUri) ? SERVER_URI : currentUri,
+                        livingMembers,
+                        enableHa,
+                        retryIntervalMs,
+                        retryTimeoutMs);
+    }
 
-	/**
-	 * List all registered listener events in Notification Service.
-	 *
-	 * @param startTime (Optional) The event create time after the given startTime.
-	 * @param startVersion (Optional) Start version of event for listening.
-	 * @param endVersion (Optional) End version of event for listening.
-	 * @return List of event updated in Notification Service.
-	 */
-	public List<EventMeta> listAllEvents(long startTime, long startVersion, long endVersion)
-			throws Exception {
-		ListAllEventsRequest request =
-				ListAllEventsRequest.newBuilder()
-						.setStartTime(startTime)
-						.setStartVersion(startVersion)
-						.setEndVersion(endVersion)
-						.build();
-		ListEventsResponse response = this.notificationServiceStub.listAllEvents(request);
-		return parseEventsFromResponse(response);
-	}
+    /** Select a valid server from server candidates as current server. */
+    protected void selectValidServer() {
+        boolean lastError = false;
+        for (MemberProto livingMember : livingMembers) {
+            try {
+                currentUri = livingMember.getServerUri();
+                initNotificationServiceStub();
+                ListMembersRequest request =
+                        ListMembersRequest.newBuilder()
+                                .setTimeoutSeconds(listMemberIntervalMs / 1000)
+                                .build();
+                ListMembersResponse response = notificationServiceStub.listMembers(request);
+                if (response.getReturnCode() == ReturnStatus.SUCCESS) {
+                    livingMembers = new HashSet<>(response.getMembersList());
+                    lastError = false;
+                    break;
+                } else {
+                    lastError = true;
+                }
+            } catch (Exception e) {
+                lastError = true;
+            }
+        }
+        if (lastError) {
+            logger.warn("No available server uri!");
+        }
+    }
 
-	private static List<EventMeta> parseEventsFromResponse(ListEventsResponse response)
-			throws Exception {
-		if (response.getReturnCode() == ReturnStatus.SUCCESS) {
-			List<EventMeta> eventMetas = new ArrayList<>();
-			for (EventProto eventProto : response.getEventsList()) {
-				eventMetas.add(buildEventMeta(eventProto));
-			}
-			return eventMetas;
-		} else {
-			throw new Exception(response.getReturnMsg());
-		}
-	}
+    /** List living members under high available mode. */
+    protected Runnable listMembers() {
+        return () -> {
+            while (enableHa) {
+                try {
+                    if (Thread.currentThread().isInterrupted()) {
+                        break;
+                    }
+                    ListMembersRequest request =
+                            ListMembersRequest.newBuilder()
+                                    .setTimeoutSeconds(listMemberIntervalMs / 1000)
+                                    .build();
+                    ListMembersResponse response = notificationServiceStub.listMembers(request);
+                    if (response.getReturnCode() == ReturnStatus.SUCCESS) {
+                        livingMembers = new HashSet<>(response.getMembersList());
+                    } else {
+                        logger.warn(response.getReturnMsg());
+                        selectValidServer();
+                    }
+                } catch (Exception e) {
+                    e.printStackTrace();
+                    logger.warn("Error while listening notification");
+                    selectValidServer();
+                }
+            }
+        };
+    }
 
-	/**
-	 * Start listen specific `key` or `version` notifications in Notification Service.
-	 *
-	 * @param key Key of notification for listening.
-	 * @param watcher Watcher instance for listening notification.
-	 * @param version (Optional) Version of notification for listening.
-	 * @param eventType (Optional) Type of event for listening.
-	 * @param startTime (Optional) Type of event for listening.
-	 */
-	public void startListenEvent(String key, EventWatcher watcher, long version, String eventType, long startTime) {
-		String namespace = this.defaultNamespace;
-		Map<String, String> listenKey = new HashMap<String, String>() {{
-			put(key, namespace);
-		}};
-		if (!this.threads.containsKey(listenKey)) {
-			ArrayList<String> curListenerKeys = new ArrayList<String>(){{add(key);}};
-			EventListener listener = new EventListener(this.notificationServiceStub,
-					curListenerKeys, version, eventType, startTime, namespace, watcher, 5);
-			listener.start();
-			this.threads.put(listenKey, listener);
-		}
-	}
+    /** Disable high availability mode. */
+    public void disableHighAvailability() {
+        enableHa = false;
+        listMembersService.shutdown();
+    }
 
-	/**
-	 * Stop listen specific `key` notifications in Notification Service.
-	 *
-	 * @param key Key of notification for listening.
-	 */
-	public void stopListenEvent(String key) {
-		String namespace = this.defaultNamespace;
-		Map<String, String> listenKey = new HashMap<String, String>(){{put(key, namespace);}};
-		if (StringUtils.isEmpty(key)) {
-			for (Map.Entry<Map<String, String>, EventListener> entry : threads.entrySet()) {
-				entry.getValue().shutdown();
-			}
-		} else {
-			if(this.threads.containsKey(listenKey)) {
-				this.threads.get(listenKey).shutdown();
-			}
-		}
-	}
+    /**
+     * Send the event to Notification Service.
+     *
+     * @param key Key of event updated in Notification Service.
+     * @param value Value of event updated in Notification Service.
+     * @param eventType Type of event updated in Notification Service.
+     * @param context Context of event updated in Notification Service.
+     * @return Object of Event created in Notification Service.
+     */
+    public EventMeta sendEvent(String key, String value, String eventType, String context)
+            throws Exception {
+        SendEventRequest request =
+                SendEventRequest.newBuilder()
+                        .setEvent(
+                                EventProto.newBuilder()
+                                        .setKey(key)
+                                        .setValue(value)
+                                        .setEventType(eventType)
+                                        .setContext(context)
+                                        .setNamespace(defaultNamespace)
+                                        .build())
+                        .setUuid(UUID.randomUUID().toString())
+                        .build();
+        SendEventsResponse response = notificationServiceStub.sendEvent(request);
+        if (response.getReturnCode() == ReturnStatus.SUCCESS) {
+            return buildEventMeta(response.getEvent());
+        } else {
+            throw new Exception(response.getReturnMsg());
+        }
+    }
 
-	/**
-	 * Get latest version of specific `key` notifications in Notification Service.
-	 *
-	 * @param key Key of notification for listening.
-	 */
-	public long getLatestVersion(String key) throws Exception {
-		if (StringUtils.isEmpty(key)) {
-			throw new Exception("Empty key, please provide valid key");
-		} else {
-			GetLatestVersionByKeyRequest request =
-					GetLatestVersionByKeyRequest.newBuilder().setKey(key).build();
-			GetLatestVersionResponse response =
-					this.notificationServiceStub.getLatestVersionByKey(request);
-			return parseLatestVersionFromResponse(response);
-		}
-	}
+    /**
+     * List specific `key` or `version` notifications in Notification Service.
+     *
+     * @param keys Keys of notification for listening.
+     * @param version (Optional) Version of notification for listening.
+     * @param eventType (Optional) Type of event for listening.
+     * @param startTime (Optional) Type of event for listening.
+     * @return List of Notification updated in Notification Service.
+     */
+    public List<EventMeta> listEvents(
+            List<String> keys, long version, String eventType, long startTime) throws Exception {
+        return listEvents(
+                notificationServiceStub, keys, version, eventType, startTime, defaultNamespace, 0);
+    }
 
+    /**
+     * List all registered listener events in Notification Service.
+     *
+     * @param startTime (Optional) The event create time after the given startTime.
+     * @param startVersion (Optional) Start version of event for listening.
+     * @param endVersion (Optional) End version of event for listening.
+     * @return List of event updated in Notification Service.
+     */
+    public List<EventMeta> listAllEvents(long startTime, long startVersion, long endVersion)
+            throws Exception {
+        ListAllEventsRequest request =
+                ListAllEventsRequest.newBuilder()
+                        .setStartTime(startTime)
+                        .setStartVersion(startVersion)
+                        .setEndVersion(endVersion)
+                        .build();
+        ListEventsResponse response = notificationServiceStub.listAllEvents(request);
+        return parseEventsFromResponse(response);
+    }
 
-	public long parseLatestVersionFromResponse(GetLatestVersionResponse response)
-			throws Exception {
-		if (response.getReturnCode().equals(ReturnStatus.ERROR.toString())) {
-			throw new Exception(response.getReturnMsg());
-		} else {
-			return response.getVersion();
-		}
-	}
+    /**
+     * Start listen specific `key` or `version` notifications in Notification Service.
+     *
+     * @param key Key of notification for listening.
+     * @param watcher Watcher instance for listening notification.
+     * @param version (Optional) Version of notification for listening.
+     * @param eventType (Optional) Type of event for listening.
+     * @param startTime (Optional) Type of event for listening.
+     */
+    public void startListenEvent(
+            String key, EventWatcher watcher, long version, String eventType, long startTime) {
+        String namespace = defaultNamespace;
+        Map<String, String> listenKey =
+                new HashMap<String, String>() {
+                    {
+                        put(key, namespace);
+                    }
+                };
+        if (!threads.containsKey(listenKey)) {
+            ArrayList<String> curListenerKeys =
+                    new ArrayList<String>() {
+                        {
+                            add(key);
+                        }
+                    };
+            EventListener listener =
+                    new EventListener(
+                            notificationServiceStub,
+                            curListenerKeys,
+                            version,
+                            eventType,
+                            startTime,
+                            namespace,
+                            watcher,
+                            5);
+            listener.start();
+            threads.put(listenKey, listener);
+        }
+    }
 
+    /**
+     * Stop listen specific `key` notifications in Notification Service.
+     *
+     * @param key Key of notification for listening.
+     */
+    public void stopListenEvent(String key) {
+        String namespace = defaultNamespace;
+        Map<String, String> listenKey =
+                new HashMap<String, String>() {
+                    {
+                        put(key, namespace);
+                    }
+                };
+        if (StringUtils.isEmpty(key)) {
+            for (Map.Entry<Map<String, String>, EventListener> entry : threads.entrySet()) {
+                entry.getValue().shutdown();
+            }
+        } else {
+            if (threads.containsKey(listenKey)) {
+                threads.get(listenKey).shutdown();
+            }
+        }
+    }
+
+    /**
+     * Get latest version of specific `key` notifications in Notification Service.
+     *
+     * @param key Key of notification for listening.
+     */
+    public long getLatestVersion(String key) throws Exception {
+        if (StringUtils.isEmpty(key)) {
+            throw new Exception("Empty key, please provide valid key");
+        } else {
+            GetLatestVersionByKeyRequest request =
+                    GetLatestVersionByKeyRequest.newBuilder().setKey(key).build();
+            GetLatestVersionResponse response =
+                    notificationServiceStub.getLatestVersionByKey(request);
+            return parseLatestVersionFromResponse(response);
+        }
+    }
+
+    public long parseLatestVersionFromResponse(GetLatestVersionResponse response) throws Exception {
+        if (response.getReturnCode().equals(ReturnStatus.ERROR.toString())) {
+            throw new Exception(response.getReturnMsg());
+        } else {
+            return response.getVersion();
+        }
+    }
 }
diff --git a/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationInterceptor.java b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationInterceptor.java
new file mode 100644
index 000000000..fe6e37745
--- /dev/null
+++ b/flink-ai-flow/lib/notification_service/java/src/main/java/com/aiflow/notification/client/NotificationInterceptor.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package com.aiflow.notification.client;
+
+import com.aiflow.notification.proto.NotificationServiceGrpc.NotificationServiceBlockingStub;
+import com.aiflow.notification.proto.NotificationServiceOuterClass.MemberProto;
+import io.grpc.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static com.aiflow.notification.client.NotificationClient.wrapBlockingStub;
+
+public class NotificationInterceptor implements ClientInterceptor {
+
+    private static final Logger logger = LoggerFactory.getLogger(NotificationInterceptor.class);
+    private NotificationServiceBlockingStub stub;
+    private String target;
+    private Set<MemberProto> livingMembers;
+    private Boolean haRunning;
+    private Integer retryIntervalMs;
+    private Integer retryTimeoutMs;
+
+    public NotificationInterceptor(
+            NotificationServiceBlockingStub stub,
+            String target,
+            Set<MemberProto> livingMembers,
+            Boolean haRunning,
+            Integer retryIntervalMs,
+            Integer retryTimeoutMs) {
+        this.stub = stub;
+        this.target = target;
+        this.livingMembers = livingMembers;
+        this.haRunning = haRunning;
+        this.retryIntervalMs = retryIntervalMs;
+        this.retryTimeoutMs = retryTimeoutMs;
+    }
+
+    @Override
+    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+            MethodDescriptor<ReqT, RespT> methodDescriptor,
+            CallOptions callOptions,
+            Channel channel) {
+        final Long startTime = System.currentTimeMillis();
+        final Set<String> failedMembers = new HashSet<>();
+        while (true) {
+            try {
+                return channel.newCall(methodDescriptor, callOptions);
+            } catch (Exception e) {
+                failedMembers.add(target);
+                boolean foundNewMember = false;
+                for (MemberProto livingMember : livingMembers) {
+                    if (foundNewMember) break;
+                    if (failedMembers.contains(livingMember.getServerUri())) continue;
+                    stub =
+                            wrapBlockingStub(
+                                    stub,
+                                    livingMember.getServerUri(),
+                                    livingMembers,
+                                    haRunning,
+                                    retryIntervalMs,
+                                    retryTimeoutMs);
+                    target = livingMember.getServerUri();
+                    foundNewMember = true;
+                }
+                if (!foundNewMember) {
+                    failedMembers.clear();
+                }
+            }
+            if (!haRunning || System.currentTimeMillis() > startTime + retryTimeoutMs) {
+                if (!haRunning) {
+                    logger.warn("HA has been disabled.");
+                } else {
+                    logger.warn("Rpc retry timeout!");
+                }
+            }
+        }
+    }
+}


Test patch:
diff --git a/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java b/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java
index 543fdb06d..ff81d7573 100644
--- a/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java
+++ b/flink-ai-flow/lib/notification_service/java/src/test/java/com/aiflow/notification/client/NotificationClientTest.java
@@ -29,95 +29,107 @@
 import org.junit.Rule;
 import org.junit.Test;
 
-import java.io.IOException;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
 
 public class NotificationClientTest {
-	private NotificationClient client;
-
-	@Rule
-	public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
-	private final MutableHandlerRegistry serviceRegistry = new MutableHandlerRegistry();
-	private LocalNotificationService mockNotificationService = new LocalNotificationService();
-
-	@Before
-	public void setUp() throws IOException {
-		// Generate a unique in-process server name.
-		String serverName = InProcessServerBuilder.generateName();
-
-		// Create a server, add service, start, and register for automatic graceful shutdown.
-		this.grpcCleanup.register(InProcessServerBuilder
-				.forName(serverName).directExecutor().fallbackHandlerRegistry(this.serviceRegistry)
-				.addService(mockNotificationService).build().start());
-
-		// Create a client channel and register for automatic graceful shutdown.
-		ManagedChannel channel = this.grpcCleanup.register(
-				InProcessChannelBuilder.forName(serverName).directExecutor().build());
-
-		// Create a NotificationClient using the in-process channel
-		this.client = new NotificationClient("localhost:50051", "default", false,
-				5, 10, 2000);
-	}
-
-	@Test
-	public void sendEvent() throws Exception {
-		long latestVersion = this.client.getLatestVersion("key");
-		for(int i = 0; i <3; i++) {
-			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
-		}
-		ArrayList<String> listenerKeys = new ArrayList<String>(){{add("key");}};
-		long startTime = System.currentTimeMillis();
-		List<EventMeta> eventList = this.client.listEvents(listenerKeys, latestVersion, "type", 0);
-		assertEquals(3, eventList.size());
-	}
-
-	@Test
-	public void listAllEvents() throws Exception{
-		long startTime=0;
-		for(int i = 0; i <3; i++) {
-			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
-			if(i == 1){
-				startTime = event.getCreateTime();
-			}
-		}
-		List<EventMeta> eventList = this.client.listAllEvents(startTime, 0, 0);
-		assertEquals(2, eventList.size());
-	}
-
-	@Test
-	public void startListenEvent() throws Exception{
-		long latestVersion = this.client.getLatestVersion("key");
-		for(int i = 0; i <3; i++) {
-			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
-
-		}
-		final Integer[] ii = { 0 };
-		String listenerKey = "key";
-		this.client.startListenEvent(listenerKey, new EventWatcher() {
-			@Override
-			public void process(List<EventMeta> events) {
-				ii[0] += events.size();
-			}
-		}, latestVersion, "type", 0);
-
-		Thread.sleep(10000);
-		this.client.stopListenEvent("key");
-		assertEquals(3, ii[0].intValue());
-	}
-
-	@Test
-	public void getLatestVersion() throws Exception {
-		long latestVersion = this.client.getLatestVersion("key");
-		for(int i = 0; i <3; i++) {
-			EventMeta event =this.client.sendEvent("key", String.valueOf(i), "type", "");
-
-		}
-		long newLatestVersion = this.client.getLatestVersion("key");
-		assertEquals(latestVersion + 3, newLatestVersion);
-	}
+    @Rule public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
+    private final MutableHandlerRegistry serviceRegistry = new MutableHandlerRegistry();
+    private NotificationClient client;
+    private LocalNotificationService mockNotificationService = new LocalNotificationService();
+
+    @Before
+    public void setUp() throws Exception {
+        // Generate a unique in-process server name.
+        String serverName = InProcessServerBuilder.generateName();
+
+        // Create a server, add service, start, and register for automatic graceful shutdown.
+        this.grpcCleanup.register(
+                InProcessServerBuilder.forName(serverName)
+                        .directExecutor()
+                        .fallbackHandlerRegistry(this.serviceRegistry)
+                        .addService(mockNotificationService)
+                        .build()
+                        .start());
+
+        // Create a client channel and register for automatic graceful shutdown.
+        ManagedChannel channel =
+                this.grpcCleanup.register(
+                        InProcessChannelBuilder.forName(serverName).directExecutor().build());
+
+        // Create a NotificationClient using the in-process channel
+        try {
+            this.client =
+                    new NotificationClient(
+                            "localhost:50051,localhost:50052", "default", true, 5, 10, 2000);
+        } catch (Exception e) {
+            throw new Exception("Failed to init notification client", e);
+        }
+    }
+
+    @Test
+    public void sendEvent() throws Exception {
+        long latestVersion = this.client.getLatestVersion("key");
+        for (int i = 0; i < 3; i++) {
+            EventMeta event = this.client.sendEvent("key", String.valueOf(i), "type", "");
+        }
+        ArrayList<String> listenerKeys =
+                new ArrayList<String>() {
+                    {
+                        add("key");
+                    }
+                };
+        long startTime = System.currentTimeMillis();
+        List<EventMeta> eventList = this.client.listEvents(listenerKeys, latestVersion, "type", 0);
+        assertEquals(3, eventList.size());
+    }
+
+    @Test
+    public void listAllEvents() throws Exception {
+        long startTime = 0;
+        for (int i = 0; i < 3; i++) {
+            EventMeta event = this.client.sendEvent("key", String.valueOf(i), "type", "");
+            if (i == 1) {
+                startTime = event.getCreateTime();
+            }
+        }
+        List<EventMeta> eventList = this.client.listAllEvents(startTime, 0, 0);
+        assertEquals(2, eventList.size());
+    }
+
+    @Test
+    public void startListenEvent() throws Exception {
+        long latestVersion = this.client.getLatestVersion("key");
+        for (int i = 0; i < 3; i++) {
+            EventMeta event = this.client.sendEvent("key", String.valueOf(i), "type", "");
+        }
+        final Integer[] ii = {0};
+        String listenerKey = "key";
+        this.client.startListenEvent(
+                listenerKey,
+                new EventWatcher() {
+                    @Override
+                    public void process(List<EventMeta> events) {
+                        ii[0] += events.size();
+                    }
+                },
+                latestVersion,
+                "type",
+                0);
+
+        Thread.sleep(10000);
+        assertEquals(3, ii[0].intValue());
+    }
+
+    @Test
+    public void getLatestVersion() throws Exception {
+        long latestVersion = this.client.getLatestVersion("key");
+        for (int i = 0; i < 3; i++) {
+            EventMeta event = this.client.sendEvent("key", String.valueOf(i), "type", "");
+        }
+        long newLatestVersion = this.client.getLatestVersion("key");
+        assertEquals(latestVersion + 3, newLatestVersion);
+    }
 }