Repository: NetEase/arctic
Pull request #126 (2022-08-04 06:48:15)

[ARCTIC-117][AMS]Ams high avaliable
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
implement https://github.com/NetEase/arctic/issues/117

## Brief change log

1.add HA config and server will register node info in zookeeper, and there is just one leader ams server provide service. when leader node down, there will reselect leader.
2.client will acquire leader node info from zookeeper, and when server reselect leader the cached client will be clear and return new client use new leader node info.

## How was this patch tested?
1.deploy zookeeper server
2.add HA config in ams server
3.start several ams server
4.start client and simulate ams server leader node down.
5.check whether client can call server. 

## Documentation

  - Does this pull request introduces a new feature? (no)
  - If yes, how is the feature documented? ( not documented)


wangtaohz: Is thrift url like `thrift://host_1:port_1,host_2:port_2,host_3:port_3/catalog` supported ?
hameizi: > Is thrift url like `thrift://host_1:port_1,host_2:port_2,host_3:port_3/catalog` supported ?

@wangtaohz not support
hameizi: > com.netease.arctic.optimizer.util.OptimizeManagerClients can be removed？

yes
hameizi: @YesOrNo828 all comment fixed in new commit. please take a look

Patch:
diff --git a/ams/ams-api/pom.xml b/ams/ams-api/pom.xml
index 04df40076..4d81106e8 100644
--- a/ams/ams-api/pom.xml
+++ b/ams/ams-api/pom.xml
@@ -55,6 +55,44 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-pool2</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+            <version>2.8.4</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-framework</artifactId>
+            <version>${curator.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-recipes</artifactId>
+            <version>${curator.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>${curator.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <version>${zookeeper.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>${fastjson.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.4</version>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
     <build>
@@ -99,6 +137,32 @@
                 <!--                    </execution>-->
                 <!--                </executions>-->
             </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <relocations>
+                                <relocation>
+                                    <pattern>org.apache.zookeeper</pattern>
+                                    <shadedPattern>com.netease.arctic.shade.org.apache.zookeeper</shadedPattern>
+                                </relocation>
+                                <relocation>
+                                    <pattern>org.apache.curator</pattern>
+                                    <shadedPattern>com.netease.arctic.shade.org.apache.curator</shadedPattern>
+                                </relocation>
+                            </relocations>
+                        </configuration>
+                    </execution>
+                </executions>
+
+            </plugin>
         </plugins>
     </build>
 
diff --git a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java
index b3c9eec94..2f3ae2a62 100644
--- a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java
+++ b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java
@@ -7,7 +7,7 @@
 package com.netease.arctic.ams.api;
 
 @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-08-22")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-08-25")
 public class OptimizeManager {
 
   /**
@@ -16,6 +16,8 @@ public class OptimizeManager {
    */
   public interface Iface {
 
+    public void ping() throws org.apache.thrift.TException;
+
     public OptimizeTask pollTask(int queueId, JobId jobId, java.lang.String attemptId, long waitTime) throws com.netease.arctic.ams.api.NoSuchObjectException, org.apache.thrift.TException;
 
     public void reportOptimizeResult(OptimizeTaskStat optimizeTaskStat) throws org.apache.thrift.TException;
@@ -26,6 +28,8 @@ public class OptimizeManager {
 
   public interface AsyncIface {
 
+    public void ping(org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
+
     public void pollTask(int queueId, JobId jobId, java.lang.String attemptId, long waitTime, org.apache.thrift.async.AsyncMethodCallback<OptimizeTask> resultHandler) throws org.apache.thrift.TException;
 
     public void reportOptimizeResult(OptimizeTaskStat optimizeTaskStat, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
@@ -54,6 +58,25 @@ public class OptimizeManager {
       super(iprot, oprot);
     }
 
+    public void ping() throws org.apache.thrift.TException
+    {
+      send_ping();
+      recv_ping();
+    }
+
+    public void send_ping() throws org.apache.thrift.TException
+    {
+      ping_args args = new ping_args();
+      sendBase("ping", args);
+    }
+
+    public void recv_ping() throws org.apache.thrift.TException
+    {
+      ping_result result = new ping_result();
+      receiveBase(result, "ping");
+      return;
+    }
+
     public OptimizeTask pollTask(int queueId, JobId jobId, java.lang.String attemptId, long waitTime) throws com.netease.arctic.ams.api.NoSuchObjectException, org.apache.thrift.TException
     {
       send_pollTask(queueId, jobId, attemptId, waitTime);
@@ -141,6 +164,35 @@ public class OptimizeManager {
       super(protocolFactory, clientManager, transport);
     }
 
+    public void ping(org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException {
+      checkReady();
+      ping_call method_call = new ping_call(resultHandler, this, ___protocolFactory, ___transport);
+      this.___currentMethod = method_call;
+      ___manager.call(method_call);
+    }
+
+    public static class ping_call extends org.apache.thrift.async.TAsyncMethodCall<Void> {
+      public ping_call(org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
+        super(client, protocolFactory, transport, resultHandler, false);
+      }
+
+      public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
+        prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("ping", org.apache.thrift.protocol.TMessageType.CALL, 0));
+        ping_args args = new ping_args();
+        args.write(prot);
+        prot.writeMessageEnd();
+      }
+
+      public Void getResult() throws org.apache.thrift.TException {
+        if (getState() != org.apache.thrift.async.TAsyncMethodCall.State.RESPONSE_READ) {
+          throw new java.lang.IllegalStateException("Method call not finished!");
+        }
+        org.apache.thrift.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift.transport.TMemoryInputTransport(getFrameBuffer().array());
+        org.apache.thrift.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport);
+        return null;
+      }
+    }
+
     public void pollTask(int queueId, JobId jobId, java.lang.String attemptId, long waitTime, org.apache.thrift.async.AsyncMethodCallback<OptimizeTask> resultHandler) throws org.apache.thrift.TException {
       checkReady();
       pollTask_call method_call = new pollTask_call(queueId, jobId, attemptId, waitTime, resultHandler, this, ___protocolFactory, ___transport);
@@ -259,12 +311,38 @@ public class OptimizeManager {
     }
 
     private static <I extends Iface> java.util.Map<java.lang.String,  org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>> getProcessMap(java.util.Map<java.lang.String, org.apache.thrift.ProcessFunction<I, ? extends  org.apache.thrift.TBase>> processMap) {
+      processMap.put("ping", new ping());
       processMap.put("pollTask", new pollTask());
       processMap.put("reportOptimizeResult", new reportOptimizeResult());
       processMap.put("reportOptimizerState", new reportOptimizerState());
       return processMap;
     }
 
+    public static class ping<I extends Iface> extends org.apache.thrift.ProcessFunction<I, ping_args> {
+      public ping() {
+        super("ping");
+      }
+
+      public ping_args getEmptyArgsInstance() {
+        return new ping_args();
+      }
+
+      protected boolean isOneway() {
+        return false;
+      }
+
+      @Override
+      protected boolean rethrowUnhandledExceptions() {
+        return false;
+      }
+
+      public ping_result getResult(I iface, ping_args args) throws org.apache.thrift.TException {
+        ping_result result = new ping_result();
+        iface.ping();
+        return result;
+      }
+    }
+
     public static class pollTask<I extends Iface> extends org.apache.thrift.ProcessFunction<I, pollTask_args> {
       public pollTask() {
         super("pollTask");
@@ -357,12 +435,73 @@ public class OptimizeManager {
     }
 
     private static <I extends AsyncIface> java.util.Map<java.lang.String,  org.apache.thrift.AsyncProcessFunction<I, ? extends  org.apache.thrift.TBase,?>> getProcessMap(java.util.Map<java.lang.String,  org.apache.thrift.AsyncProcessFunction<I, ? extends  org.apache.thrift.TBase, ?>> processMap) {
+      processMap.put("ping", new ping());
       processMap.put("pollTask", new pollTask());
       processMap.put("reportOptimizeResult", new reportOptimizeResult());
       processMap.put("reportOptimizerState", new reportOptimizerState());
       return processMap;
     }
 
+    public static class ping<I extends AsyncIface> extends org.apache.thrift.AsyncProcessFunction<I, ping_args, Void> {
+      public ping() {
+        super("ping");
+      }
+
+      public ping_args getEmptyArgsInstance() {
+        return new ping_args();
+      }
+
+      public org.apache.thrift.async.AsyncMethodCallback<Void> getResultHandler(final org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer fb, final int seqid) {
+        final org.apache.thrift.AsyncProcessFunction fcall = this;
+        return new org.apache.thrift.async.AsyncMethodCallback<Void>() { 
+          public void onComplete(Void o) {
+            ping_result result = new ping_result();
+            try {
+              fcall.sendResponse(fb, result, org.apache.thrift.protocol.TMessageType.REPLY,seqid);
+            } catch (org.apache.thrift.transport.TTransportException e) {
+              _LOGGER.error("TTransportException writing to internal frame buffer", e);
+              fb.close();
+            } catch (java.lang.Exception e) {
+              _LOGGER.error("Exception writing to internal frame buffer", e);
+              onError(e);
+            }
+          }
+          public void onError(java.lang.Exception e) {
+            byte msgType = org.apache.thrift.protocol.TMessageType.REPLY;
+            org.apache.thrift.TSerializable msg;
+            ping_result result = new ping_result();
+            if (e instanceof org.apache.thrift.transport.TTransportException) {
+              _LOGGER.error("TTransportException inside handler", e);
+              fb.close();
+              return;
+            } else if (e instanceof org.apache.thrift.TApplicationException) {
+              _LOGGER.error("TApplicationException inside handler", e);
+              msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
+              msg = (org.apache.thrift.TApplicationException)e;
+            } else {
+              _LOGGER.error("Exception inside handler", e);
+              msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
+              msg = new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.INTERNAL_ERROR, e.getMessage());
+            }
+            try {
+              fcall.sendResponse(fb,msg,msgType,seqid);
+            } catch (java.lang.Exception ex) {
+              _LOGGER.error("Exception writing to internal frame buffer", ex);
+              fb.close();
+            }
+          }
+        };
+      }
+
+      protected boolean isOneway() {
+        return false;
+      }
+
+      public void start(I iface, ping_args args, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException {
+        iface.ping(resultHandler);
+      }
+    }
+
     public static class pollTask<I extends AsyncIface> extends org.apache.thrift.AsyncProcessFunction<I, pollTask_args, OptimizeTask> {
       public pollTask() {
         super("pollTask");
@@ -550,6 +689,514 @@ public class OptimizeManager {
 
   }
 
+  public static class ping_args implements org.apache.thrift.TBase<ping_args, ping_args._Fields>, java.io.Serializable, Cloneable, Comparable<ping_args>   {
+    private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ping_args");
+
+
+    private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new ping_argsStandardSchemeFactory();
+    private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new ping_argsTupleSchemeFactory();
+
+
+    /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+    public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+;
+
+      private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+      static {
+        for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+          byName.put(field.getFieldName(), field);
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, or null if its not found.
+       */
+      @org.apache.thrift.annotation.Nullable
+      public static _Fields findByThriftId(int fieldId) {
+        switch(fieldId) {
+          default:
+            return null;
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, throwing an exception
+       * if it is not found.
+       */
+      public static _Fields findByThriftIdOrThrow(int fieldId) {
+        _Fields fields = findByThriftId(fieldId);
+        if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+        return fields;
+      }
+
+      /**
+       * Find the _Fields constant that matches name, or null if its not found.
+       */
+      @org.apache.thrift.annotation.Nullable
+      public static _Fields findByName(java.lang.String name) {
+        return byName.get(name);
+      }
+
+      private final short _thriftId;
+      private final java.lang.String _fieldName;
+
+      _Fields(short thriftId, java.lang.String fieldName) {
+        _thriftId = thriftId;
+        _fieldName = fieldName;
+      }
+
+      public short getThriftFieldId() {
+        return _thriftId;
+      }
+
+      public java.lang.String getFieldName() {
+        return _fieldName;
+      }
+    }
+    public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+    static {
+      java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+      metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+      org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(ping_args.class, metaDataMap);
+    }
+
+    public ping_args() {
+    }
+
+    /**
+     * Performs a deep copy on <i>other</i>.
+     */
+    public ping_args(ping_args other) {
+    }
+
+    public ping_args deepCopy() {
+      return new ping_args(this);
+    }
+
+    @Override
+    public void clear() {
+    }
+
+    public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+      switch (field) {
+      }
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public java.lang.Object getFieldValue(_Fields field) {
+      switch (field) {
+      }
+      throw new java.lang.IllegalStateException();
+    }
+
+    /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+    public boolean isSet(_Fields field) {
+      if (field == null) {
+        throw new java.lang.IllegalArgumentException();
+      }
+
+      switch (field) {
+      }
+      throw new java.lang.IllegalStateException();
+    }
+
+    @Override
+    public boolean equals(java.lang.Object that) {
+      if (that == null)
+        return false;
+      if (that instanceof ping_args)
+        return this.equals((ping_args)that);
+      return false;
+    }
+
+    public boolean equals(ping_args that) {
+      if (that == null)
+        return false;
+      if (this == that)
+        return true;
+
+      return true;
+    }
+
+    @Override
+    public int hashCode() {
+      int hashCode = 1;
+
+      return hashCode;
+    }
+
+    @Override
+    public int compareTo(ping_args other) {
+      if (!getClass().equals(other.getClass())) {
+        return getClass().getName().compareTo(other.getClass().getName());
+      }
+
+      int lastComparison = 0;
+
+      return 0;
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public _Fields fieldForId(int fieldId) {
+      return _Fields.findByThriftId(fieldId);
+    }
+
+    public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+      scheme(iprot).read(iprot, this);
+    }
+
+    public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+      scheme(oprot).write(oprot, this);
+    }
+
+    @Override
+    public java.lang.String toString() {
+      java.lang.StringBuilder sb = new java.lang.StringBuilder("ping_args(");
+      boolean first = true;
+
+      sb.append(")");
+      return sb.toString();
+    }
+
+    public void validate() throws org.apache.thrift.TException {
+      // check for required fields
+      // check for sub-struct validity
+    }
+
+    private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+      try {
+        write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+      try {
+        read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private static class ping_argsStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public ping_argsStandardScheme getScheme() {
+        return new ping_argsStandardScheme();
+      }
+    }
+
+    private static class ping_argsStandardScheme extends org.apache.thrift.scheme.StandardScheme<ping_args> {
+
+      public void read(org.apache.thrift.protocol.TProtocol iprot, ping_args struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TField schemeField;
+        iprot.readStructBegin();
+        while (true)
+        {
+          schemeField = iprot.readFieldBegin();
+          if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
+            break;
+          }
+          switch (schemeField.id) {
+            default:
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+          }
+          iprot.readFieldEnd();
+        }
+        iprot.readStructEnd();
+
+        // check for required fields of primitive type, which can't be checked in the validate method
+        struct.validate();
+      }
+
+      public void write(org.apache.thrift.protocol.TProtocol oprot, ping_args struct) throws org.apache.thrift.TException {
+        struct.validate();
+
+        oprot.writeStructBegin(STRUCT_DESC);
+        oprot.writeFieldStop();
+        oprot.writeStructEnd();
+      }
+
+    }
+
+    private static class ping_argsTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public ping_argsTupleScheme getScheme() {
+        return new ping_argsTupleScheme();
+      }
+    }
+
+    private static class ping_argsTupleScheme extends org.apache.thrift.scheme.TupleScheme<ping_args> {
+
+      @Override
+      public void write(org.apache.thrift.protocol.TProtocol prot, ping_args struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      }
+
+      @Override
+      public void read(org.apache.thrift.protocol.TProtocol prot, ping_args struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      }
+    }
+
+    private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+      return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+    }
+  }
+
+  public static class ping_result implements org.apache.thrift.TBase<ping_result, ping_result._Fields>, java.io.Serializable, Cloneable, Comparable<ping_result>   {
+    private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ping_result");
+
+
+    private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new ping_resultStandardSchemeFactory();
+    private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new ping_resultTupleSchemeFactory();
+
+
+    /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+    public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+;
+
+      private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+      static {
+        for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+          byName.put(field.getFieldName(), field);
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, or null if its not found.
+       */
+      @org.apache.thrift.annotation.Nullable
+      public static _Fields findByThriftId(int fieldId) {
+        switch(fieldId) {
+          default:
+            return null;
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, throwing an exception
+       * if it is not found.
+       */
+      public static _Fields findByThriftIdOrThrow(int fieldId) {
+        _Fields fields = findByThriftId(fieldId);
+        if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+        return fields;
+      }
+
+      /**
+       * Find the _Fields constant that matches name, or null if its not found.
+       */
+      @org.apache.thrift.annotation.Nullable
+      public static _Fields findByName(java.lang.String name) {
+        return byName.get(name);
+      }
+
+      private final short _thriftId;
+      private final java.lang.String _fieldName;
+
+      _Fields(short thriftId, java.lang.String fieldName) {
+        _thriftId = thriftId;
+        _fieldName = fieldName;
+      }
+
+      public short getThriftFieldId() {
+        return _thriftId;
+      }
+
+      public java.lang.String getFieldName() {
+        return _fieldName;
+      }
+    }
+    public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+    static {
+      java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+      metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+      org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(ping_result.class, metaDataMap);
+    }
+
+    public ping_result() {
+    }
+
+    /**
+     * Performs a deep copy on <i>other</i>.
+     */
+    public ping_result(ping_result other) {
+    }
+
+    public ping_result deepCopy() {
+      return new ping_result(this);
+    }
+
+    @Override
+    public void clear() {
+    }
+
+    public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+      switch (field) {
+      }
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public java.lang.Object getFieldValue(_Fields field) {
+      switch (field) {
+      }
+      throw new java.lang.IllegalStateException();
+    }
+
+    /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+    public boolean isSet(_Fields field) {
+      if (field == null) {
+        throw new java.lang.IllegalArgumentException();
+      }
+
+      switch (field) {
+      }
+      throw new java.lang.IllegalStateException();
+    }
+
+    @Override
+    public boolean equals(java.lang.Object that) {
+      if (that == null)
+        return false;
+      if (that instanceof ping_result)
+        return this.equals((ping_result)that);
+      return false;
+    }
+
+    public boolean equals(ping_result that) {
+      if (that == null)
+        return false;
+      if (this == that)
+        return true;
+
+      return true;
+    }
+
+    @Override
+    public int hashCode() {
+      int hashCode = 1;
+
+      return hashCode;
+    }
+
+    @Override
+    public int compareTo(ping_result other) {
+      if (!getClass().equals(other.getClass())) {
+        return getClass().getName().compareTo(other.getClass().getName());
+      }
+
+      int lastComparison = 0;
+
+      return 0;
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public _Fields fieldForId(int fieldId) {
+      return _Fields.findByThriftId(fieldId);
+    }
+
+    public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+      scheme(iprot).read(iprot, this);
+    }
+
+    public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+      scheme(oprot).write(oprot, this);
+      }
+
+    @Override
+    public java.lang.String toString() {
+      java.lang.StringBuilder sb = new java.lang.StringBuilder("ping_result(");
+      boolean first = true;
+
+      sb.append(")");
+      return sb.toString();
+    }
+
+    public void validate() throws org.apache.thrift.TException {
+      // check for required fields
+      // check for sub-struct validity
+    }
+
+    private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+      try {
+        write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+      try {
+        read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private static class ping_resultStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public ping_resultStandardScheme getScheme() {
+        return new ping_resultStandardScheme();
+      }
+    }
+
+    private static class ping_resultStandardScheme extends org.apache.thrift.scheme.StandardScheme<ping_result> {
+
+      public void read(org.apache.thrift.protocol.TProtocol iprot, ping_result struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TField schemeField;
+        iprot.readStructBegin();
+        while (true)
+        {
+          schemeField = iprot.readFieldBegin();
+          if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
+            break;
+          }
+          switch (schemeField.id) {
+            default:
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+          }
+          iprot.readFieldEnd();
+        }
+        iprot.readStructEnd();
+
+        // check for required fields of primitive type, which can't be checked in the validate method
+        struct.validate();
+      }
+
+      public void write(org.apache.thrift.protocol.TProtocol oprot, ping_result struct) throws org.apache.thrift.TException {
+        struct.validate();
+
+        oprot.writeStructBegin(STRUCT_DESC);
+        oprot.writeFieldStop();
+        oprot.writeStructEnd();
+      }
+
+    }
+
+    private static class ping_resultTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public ping_resultTupleScheme getScheme() {
+        return new ping_resultTupleScheme();
+      }
+    }
+
+    private static class ping_resultTupleScheme extends org.apache.thrift.scheme.TupleScheme<ping_result> {
+
+      @Override
+      public void write(org.apache.thrift.protocol.TProtocol prot, ping_result struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      }
+
+      @Override
+      public void read(org.apache.thrift.protocol.TProtocol prot, ping_result struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      }
+    }
+
+    private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+      return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+    }
+  }
+
   public static class pollTask_args implements org.apache.thrift.TBase<pollTask_args, pollTask_args._Fields>, java.io.Serializable, Cloneable, Comparable<pollTask_args>   {
     private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("pollTask_args");
 
diff --git a/core/src/main/java/com/netease/arctic/AmsClientPools.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/AmsClientPools.java
similarity index 54%
rename from core/src/main/java/com/netease/arctic/AmsClientPools.java
rename to ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/AmsClientPools.java
index 4f3da6a5e..69cae82c3 100644
--- a/core/src/main/java/com/netease/arctic/AmsClientPools.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/AmsClientPools.java
@@ -1,37 +1,13 @@
-/*
- * 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.netease.arctic;
+package com.netease.arctic.ams.api.client;
 
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.netease.arctic.ams.api.ArcticTableMetastore;
-import com.netease.arctic.ams.api.client.ArcticThriftUrl;
-import com.netease.arctic.ams.api.client.PoolConfig;
-import com.netease.arctic.ams.api.client.ServiceInfo;
-import com.netease.arctic.ams.api.client.ThriftClientPool;
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TBinaryProtocol;
 import org.apache.thrift.protocol.TMultiplexedProtocol;
 import org.apache.thrift.protocol.TProtocol;
 
-import java.util.Collections;
-
 /**
  * Client pool cache for different ams server, sharing in jvm.
  */
@@ -53,14 +29,12 @@ public class AmsClientPools {
   }
 
   private static ThriftClientPool<ArcticTableMetastore.Client> buildClientPool(String url) {
-    ArcticThriftUrl arcticThriftUrl = ArcticThriftUrl.parse(url);
     PoolConfig poolConfig = new PoolConfig();
-    poolConfig.setTimeout(arcticThriftUrl.socketTimeout());
     poolConfig.setFailover(true);
     poolConfig.setMinIdle(CLIENT_POOL_MIN);
     poolConfig.setMaxIdle(CLIENT_POOL_MAX);
-    return new ThriftClientPool<>(Collections.singletonList(
-        new ServiceInfo(arcticThriftUrl.host(), arcticThriftUrl.port())),
+    return new ThriftClientPool<>(
+        url,
         s -> {
           TProtocol protocol = new TBinaryProtocol(s);
           ArcticTableMetastore.Client tableMetastore = new ArcticTableMetastore.Client(
@@ -74,6 +48,6 @@ public class AmsClientPools {
             return false;
           }
           return true;
-        }, new PoolConfig());
+        }, poolConfig);
   }
 }
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/AmsServerInfo.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/AmsServerInfo.java
new file mode 100644
index 000000000..97d17e784
--- /dev/null
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/AmsServerInfo.java
@@ -0,0 +1,41 @@
+package com.netease.arctic.ams.api.client;
+
+import java.util.Objects;
+
+public class AmsServerInfo {
+  private String host;
+  private Integer thriftBindPort;
+
+  public String getHost() {
+    return host;
+  }
+
+  public void setHost(String host) {
+    this.host = host;
+  }
+
+  public Integer getThriftBindPort() {
+    return thriftBindPort;
+  }
+
+  public void setThriftBindPort(Integer thriftBindPort) {
+    this.thriftBindPort = thriftBindPort;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    AmsServerInfo that = (AmsServerInfo) o;
+    return Objects.equals(host, that.host) && Objects.equals(thriftBindPort, that.thriftBindPort);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(host, thriftBindPort);
+  }
+}
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ArcticThriftUrl.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ArcticThriftUrl.java
index 5057e7ec3..be30487bc 100644
--- a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ArcticThriftUrl.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ArcticThriftUrl.java
@@ -18,13 +18,20 @@
 
 package com.netease.arctic.ams.api.client;
 
+import com.alibaba.fastjson.JSONObject;
+import com.netease.arctic.ams.api.properties.AmsHAProperties;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Locale;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 public class ArcticThriftUrl {
   public static final String PARAM_SOCKET_TIMEOUT = "socketTimeout";
   public static final int DEFAULT_SOCKET_TIMEOUT = 5000;
+  public static final String ZOOKEEPER_FLAG = "zookeeper";
+  public static final String THRIFT_FLAG = "thrift";
+  private static final Pattern PATTERN = Pattern.compile("zookeeper://(\\S+)/(\\w+)");
   private final String schema;
   private final String host;
   private final int port;
@@ -35,7 +42,9 @@ public class ArcticThriftUrl {
   private final String url;
 
   /**
-   * parse thrift url.
+   * parse thrift url, now support thrift://host:port/{catalogName} and zookeeper://host:port/{cluster}/{catalogName}
+   * . parse to ArcticThriftUrl that contains properties schema, host, port, catalogName, socketTimeout, url.
+   *
    * @param url - thrift url
    * @return -
    */
@@ -43,6 +52,41 @@ public class ArcticThriftUrl {
     if (url == null) {
       throw new IllegalArgumentException("thrift url is null");
     }
+    if (url.startsWith(ZOOKEEPER_FLAG)) {
+      String thriftUrl = url;
+      String query = "";
+      if (url.contains("?")) {
+        query = url.substring(url.indexOf("?"));
+        thriftUrl = url.substring(0, url.indexOf("?"));
+      }
+      Matcher m = PATTERN.matcher(thriftUrl);
+      if (m.matches()) {
+        String zkServerAddress;
+        String cluster;
+        String catalog = "";
+        if (m.group(1).contains("/")) {
+          zkServerAddress = m.group(1).substring(0, m.group(1).indexOf("/"));
+          cluster = m.group(1).substring(m.group(1).indexOf("/") + 1);
+          catalog = m.group(2);
+        } else {
+          zkServerAddress = m.group(1);
+          cluster = m.group(2);
+        }
+        ZookeeperService zkService = ZookeeperService.getInstance(zkServerAddress);
+        AmsServerInfo serverInfo = null;
+        try {
+          serverInfo = JSONObject.parseObject(
+              zkService.getData(AmsHAProperties.getMasterPath(cluster)),
+              AmsServerInfo.class);
+        } catch (Exception e) {
+          throw new RuntimeException("get master server info from zookeeper error");
+        }
+        url =
+            String.format("thrift://%s:%d/%s%s", serverInfo.getHost(), serverInfo.getThriftBindPort(), catalog, query);
+      } else {
+        throw new RuntimeException(String.format("invalid ams url %s", url));
+      }
+    }
     String schema;
     String host;
     int port;
@@ -74,8 +118,9 @@ public class ArcticThriftUrl {
     }
   }
 
-  private ArcticThriftUrl(String schema, String host, int port, String catalogName, int socketTimeout,
-                         String url) {
+  private ArcticThriftUrl(
+      String schema, String host, int port, String catalogName, int socketTimeout,
+      String url) {
     this.schema = schema;
     this.host = host;
     this.port = port;
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/util/OptimizeManagerClients.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/OptimizeManagerClientPools.java
similarity index 69%
rename from optimizer/src/main/java/com/netease/arctic/optimizer/util/OptimizeManagerClients.java
rename to ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/OptimizeManagerClientPools.java
index 4bffba82d..000395e35 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/util/OptimizeManagerClients.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/OptimizeManagerClientPools.java
@@ -16,47 +16,49 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.optimizer.util;
+package com.netease.arctic.ams.api.client;
 
 
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.netease.arctic.ams.api.ArcticTableMetastore;
 import com.netease.arctic.ams.api.OptimizeManager;
-import com.netease.arctic.ams.api.client.ArcticThriftUrl;
-import com.netease.arctic.ams.api.client.PoolConfig;
-import com.netease.arctic.ams.api.client.ServiceInfo;
-import com.netease.arctic.ams.api.client.ThriftClientPool;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TBinaryProtocol;
 import org.apache.thrift.protocol.TMultiplexedProtocol;
 
 import java.util.Collections;
 
-public class OptimizeManagerClients {
+/**
+ * Client pool cache for different ams optimize server, sharing in jvm.
+ */
+public class OptimizeManagerClientPools {
   private static final int CLIENT_POOL_MIN = 1;
   private static final int CLIENT_POOL_MAX = 5;
 
   private static final LoadingCache<String, ThriftClientPool<OptimizeManager.Client>> CLIENT_POOLS
       = Caffeine.newBuilder()
-      .build(OptimizeManagerClients::buildClient);
+      .build(OptimizeManagerClientPools::buildClient);
 
   public static OptimizeManager.Iface getClient(String metastoreUrl) {
-    Preconditions.checkNotNull(metastoreUrl, "metastoreUrl is null");
     return CLIENT_POOLS.get(metastoreUrl).iface();
   }
 
   private static ThriftClientPool<OptimizeManager.Client> buildClient(String url) {
-    ArcticThriftUrl arcticThriftUrl = ArcticThriftUrl.parse(url);
     PoolConfig poolConfig = new PoolConfig();
-    poolConfig.setTimeout(arcticThriftUrl.socketTimeout());
     poolConfig.setFailover(true);
     poolConfig.setMinIdle(CLIENT_POOL_MIN);
     poolConfig.setMaxIdle(CLIENT_POOL_MAX);
-    return new ThriftClientPool<>(Collections.singletonList(
-        new ServiceInfo(arcticThriftUrl.host(), arcticThriftUrl.port())),
+    return new ThriftClientPool<>(url,
         s -> new OptimizeManager.Client(
-            new TMultiplexedProtocol(new TBinaryProtocol(s), "OptimizeManager")), c -> true, new PoolConfig());
+            new TMultiplexedProtocol(new TBinaryProtocol(s), "OptimizeManager")), c -> {
+      try {
+        ((OptimizeManager.Client) c).ping();
+      } catch (TException e) {
+        return false;
+      }
+      return true;
+    }, poolConfig);
   }
 
 }
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ServiceInfo.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ServiceInfo.java
index 1f424595d..abae6c592 100644
--- a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ServiceInfo.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ServiceInfo.java
@@ -20,9 +20,9 @@ package com.netease.arctic.ams.api.client;
 
 public class ServiceInfo {
 
-  private final String host;
+  private String host;
 
-  private final int port;
+  private int port;
 
   public ServiceInfo(String host, int port) {
     super();
@@ -38,6 +38,14 @@ public class ServiceInfo {
     return port;
   }
 
+  public void setHost(String host) {
+    this.host = host;
+  }
+
+  public void setPort(int port) {
+    this.port = port;
+  }
+
   @Override
   public int hashCode() {
     final int prime = 31;
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ThriftClientPool.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ThriftClientPool.java
index 16ffb4e48..6229eb330 100644
--- a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ThriftClientPool.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ThriftClientPool.java
@@ -38,7 +38,7 @@ import java.util.stream.Collectors;
 
 public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
 
-  private final Logger logger = LoggerFactory.getLogger(getClass());
+  private static final Logger LOG = LoggerFactory.getLogger("ThriftClientPool");
 
   private final ThriftClientFactory clientFactory;
 
@@ -46,7 +46,7 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
 
   private final GenericObjectPool<ThriftClient<T>> pool;
 
-  private List<ServiceInfo> services;
+  private String url;
 
   private boolean serviceReset = false;
 
@@ -54,27 +54,30 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
   // for thrift connects
   private static final int retries = 5;
 
+  private static final int retryInterval = 2000;
+
   /**
    * Construct a new pool using default config
    *
-   * @param services
+   * @param url
    * @param factory
    */
-  public ThriftClientPool(List<ServiceInfo> services, ThriftClientFactory factory, ThriftPingFactory pingFactory) {
-    this(services, factory, pingFactory, new PoolConfig());
+  public ThriftClientPool(String url, ThriftClientFactory factory, ThriftPingFactory pingFactory) {
+    this(url, factory, pingFactory, new PoolConfig());
   }
 
   /**
    * Construct a new pool using
    *
-   * @param services
+   * @param url
    * @param factory
    * @param config
    */
-  public ThriftClientPool(List<ServiceInfo> services, ThriftClientFactory factory, ThriftPingFactory pingFactory,
-                          PoolConfig config) {
-    if (services == null || services.size() == 0) {
-      throw new IllegalArgumentException("services is empty!");
+  public ThriftClientPool(
+      String url, ThriftClientFactory factory, ThriftPingFactory pingFactory,
+      PoolConfig config) {
+    if (url == null || url.isEmpty()) {
+      throw new IllegalArgumentException("url is empty!");
     }
     if (factory == null) {
       throw new IllegalArgumentException("factory is empty!");
@@ -83,7 +86,7 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
       throw new IllegalArgumentException("config is empty!");
     }
 
-    this.services = services;
+    this.url = url;
     this.clientFactory = factory;
     this.pingFactory = pingFactory;
     this.poolConfig = config;
@@ -96,29 +99,36 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
       public ThriftClient<T> create() throws Exception {
 
         // get from global list first
-        List<ServiceInfo> serviceList = ThriftClientPool.this.services;
-        ServiceInfo serviceInfo = getRandomService(serviceList);
+        ArcticThriftUrl arcticThriftUrl = ArcticThriftUrl.parse(url);
+        ServiceInfo serviceInfo = new ServiceInfo(arcticThriftUrl.host(), arcticThriftUrl.port());
         TTransport transport = getTransport(serviceInfo);
 
         try {
           transport.open();
         } catch (TTransportException e) {
-          logger.info("transport open fail service: host={}, port={}",
+          LOG.warn("transport open fail service: host={}, port={}",
               serviceInfo.getHost(), serviceInfo.getPort());
           if (poolConfig.isFailover()) {
-            while (true) {
+            for (int i = 0; i < 5; i++) {
               try {
-                // mark current fail and try next, until none service available
-                serviceList = removeFailService(serviceList, serviceInfo);
-                serviceInfo = getRandomService(serviceList);
+                arcticThriftUrl = ArcticThriftUrl.parse(url);
+                serviceInfo.setHost(arcticThriftUrl.host());
+                serviceInfo.setPort(arcticThriftUrl.port());
                 transport = getTransport(serviceInfo); // while break here
-                logger.info("failover to next service host={}, port={}",
+                LOG.info("failover to next service host={}, port={}",
                     serviceInfo.getHost(), serviceInfo.getPort());
                 transport.open();
                 break;
               } catch (TTransportException e2) {
-                logger.warn("failover fail, services left: {}", serviceList.size());
+                LOG.warn("transport open fail service: host={}, port={}",
+                    serviceInfo.getHost(), serviceInfo.getPort());
               }
+              Thread.sleep(retryInterval);
+            }
+            if (!transport.isOpen()) {
+              throw new ConnectionFailException(
+                  "connect error after try 5 times, last connect is: host=" + serviceInfo.getHost() + ", ip=" +
+                      serviceInfo.getPort(), e);
             }
           } else {
             throw new ConnectionFailException("host=" + serviceInfo.getHost() + ", ip=" + serviceInfo.getPort(), e);
@@ -128,7 +138,7 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
         ThriftClient<T> client = new ThriftClient<>(clientFactory.createClient(transport),
             pool, serviceInfo);
 
-        logger.debug("create new object for pool {}", client);
+        LOG.debug("create new object for pool {}", client);
         return client;
       }
 
@@ -140,16 +150,9 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
       @Override
       public boolean validateObject(PooledObject<ThriftClient<T>> p) {
         ThriftClient<T> client = p.getObject();
-
-        // check if return client in current service list if 
-        if (serviceReset) {
-          if (!ThriftClientPool.this.services.contains(client.getServiceInfo())) {
-            logger.warn("not return object cuase it's from previous config {}", client);
-            client.closeClient();
-            return false;
-          }
+        if (client.isDisConnected() || !pingFactory.ping(client.iface())) {
+          return false;
         }
-
         return super.validateObject(p);
       }
 
@@ -161,19 +164,6 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
     }, poolConfig);
   }
 
-  /**
-   * set new services for this pool
-   *
-   * @param services
-   */
-  public void setServices(List<ServiceInfo> services) {
-    if (services == null || services.size() == 0) {
-      throw new IllegalArgumentException("services is empty!");
-    }
-    this.services = services;
-    serviceReset = true;
-  }
-
   private TTransport getTransport(ServiceInfo serviceInfo) throws TTransportException {
 
     if (serviceInfo == null) {
@@ -204,7 +194,7 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
   }
 
   private List<ServiceInfo> removeFailService(List<ServiceInfo> list, ServiceInfo serviceInfo) {
-    logger.info("remove service from current service list: host={}, port={}",
+    LOG.info("remove service from current service list: host={}, port={}",
         serviceInfo.getHost(), serviceInfo.getPort());
     return list.stream() //
         .filter(si -> !serviceInfo.equals(si)) //
@@ -242,11 +232,17 @@ public class ThriftClientPool<T extends org.apache.thrift.TServiceClient> {
       try {
         client = pool.borrowObject();
         if (client.isDisConnected() || !pingFactory.ping(client.iface())) {
-          client.close();
+          if (attempt > 1) {
+            // if attempt > 1, it means the server is maybe restarting, so we should wait a while
+            LOG.warn("server is restarting, wait a while");
+            Thread.sleep(retryInterval);
+          }
+          pool.clear();
+          client = pool.borrowObject();
         } else {
           break;
         }
-      } catch(Exception e){
+      } catch (Exception e) {
         if (e instanceof ThriftException) {
           throw (ThriftException) e;
         }
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ZookeeperService.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ZookeeperService.java
new file mode 100644
index 000000000..48d3cb2d6
--- /dev/null
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/client/ZookeeperService.java
@@ -0,0 +1,95 @@
+/*
+ * 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.netease.arctic.ams.api.client;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.data.Stat;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Provides ZooKeeper clients and operations.
+ */
+public class ZookeeperService {
+
+  private CuratorFramework zkClient;
+  private String zkServerAddress;
+  private static volatile ZookeeperService instance;
+
+  private ZookeeperService(String zkServerAddress) {
+    this.zkServerAddress = zkServerAddress;
+    this.zkClient = newClient();
+  }
+
+  private CuratorFramework newClient() {
+    ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
+    CuratorFramework client = CuratorFrameworkFactory.builder()
+        .connectString(zkServerAddress)
+        .sessionTimeoutMs(5000)
+        .connectionTimeoutMs(5000)
+        .retryPolicy(retryPolicy)
+        .build();
+    client.start();
+    return client;
+  }
+
+  public static ZookeeperService getInstance(String zkServerAddress) {
+    if (instance == null) {
+      synchronized (ZookeeperService.class) {
+        if (instance == null) {
+          instance = new ZookeeperService(zkServerAddress);
+        }
+      }
+    }
+    return instance;
+  }
+
+  public boolean exist(String path) throws Exception {
+    Stat stat = zkClient.checkExists().forPath(path);
+    return stat != null;
+  }
+
+  public void create(String path) throws Exception {
+    StringBuilder tmpPath = new StringBuilder();
+    for (String p : path.split("/")) {
+      if (!p.isEmpty()) {
+        tmpPath.append("/");
+        tmpPath.append(p);
+        if (!exist(tmpPath.toString())) {
+          zkClient.create().withMode(CreateMode.PERSISTENT).forPath(tmpPath.toString());
+        }
+      }
+    }
+  }
+
+  public void setData(String path, String data) throws Exception {
+    zkClient.setData().forPath(path, data.getBytes(StandardCharsets.UTF_8));
+  }
+
+  public String getData(String path) throws Exception {
+    return new String(zkClient.getData().forPath(path), StandardCharsets.UTF_8);
+  }
+
+  public void delete(String path) throws Exception {
+    zkClient.delete().forPath(path);
+  }
+}
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/AmsHAProperties.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/AmsHAProperties.java
new file mode 100644
index 000000000..9fda3d490
--- /dev/null
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/AmsHAProperties.java
@@ -0,0 +1,51 @@
+/*
+ * 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.netease.arctic.ams.api.properties;
+
+public class AmsHAProperties {
+  private static final String ROOT_PATH = "/arctic/ams";
+  private static final String LEADER_PATH = "/leader";
+  private static final String MASTER_PATH = "/master";
+  private static final String NAMESPACE_DEFAULT = "/master";
+
+  public static String getBasePath(String namespace) {
+    if (namespace.isEmpty()) {
+      namespace = getNamespaceDefault();
+    }
+    return "/" + namespace + ROOT_PATH;
+  }
+
+  public static String getMasterPath(String namespace) {
+    if (namespace.isEmpty()) {
+      namespace = getNamespaceDefault();
+    }
+    return "/" + namespace + ROOT_PATH + MASTER_PATH;
+  }
+
+  public static String getLeaderPath(String namespace) {
+    if (namespace.isEmpty()) {
+      namespace = getNamespaceDefault();
+    }
+    return "/" + namespace + ROOT_PATH + LEADER_PATH;
+  }
+
+  public static String getNamespaceDefault() {
+    return NAMESPACE_DEFAULT;
+  }
+}
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/OptimizerProperties.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/OptimizerProperties.java
index e55c028c9..cd64f5c94 100644
--- a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/OptimizerProperties.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/OptimizerProperties.java
@@ -34,6 +34,9 @@ public class OptimizerProperties {
   public static final String ARCTIC_HOME = "ARCTIC_HOME";
   public static final String THRIFT_BIND_HOST = "arctic.ams.server-host";
   public static final String THRIFT_BIND_PORT = "arctic.ams.thrift.port";
+  public static final String HA_ENABLE = "arctic.ams.ha.enable";
+  public static final String CLUSTER_NAME = "arctic.ams.cluster.name";
+  public static final String ZOOKEEPER_SERVER = "arctic.ams.zookeeper.server";
 
   //optimizer launcher properties
   public static final String OPTIMIZER_LAUNCHER_INFO = "launcherInfo";
diff --git a/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift b/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift
index 1cbc08f41..3cb5c6348 100644
--- a/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift
+++ b/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift
@@ -98,6 +98,8 @@ struct DataFileInfo {
 **/
 service OptimizeManager {
 
+    void ping()
+
     OptimizeTask pollTask(1:i32 queueId, 2:JobId jobId, 3:string attemptId, 4:i64 waitTime)
         throws (1: arctic_commons.NoSuchObjectException e1)
 
diff --git a/ams/ams-server/pom.xml b/ams/ams-server/pom.xml
index 632e38a2a..1cf2e9166 100644
--- a/ams/ams-server/pom.xml
+++ b/ams/ams-server/pom.xml
@@ -72,6 +72,14 @@
                     <groupId>log4j</groupId>
                     <artifactId>log4j</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-framework</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-recipes</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 <!--        <dependency>-->
@@ -133,6 +141,14 @@
                     <groupId>org.apache.avro</groupId>
                     <artifactId>avro</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-framework</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-recipes</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 
@@ -299,6 +315,18 @@
                     <groupId>org.apache.avro</groupId>
                     <artifactId>avro</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-framework</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-recipes</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.curator</groupId>
+                    <artifactId>curator-client</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 
@@ -445,6 +473,27 @@
             <artifactId>parquet-hadoop</artifactId>
             <version>${parquet.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-framework</artifactId>
+            <version>${curator.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-recipes</artifactId>
+            <version>${curator.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <version>${zookeeper.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>${curator.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -499,7 +548,7 @@
                 <configuration>
                     <includes>
                         <include>**/AmsTestBase.java</include>
-<!--                        <include>**/OutAmsTestBase.java</include>-->
+                        <include>**/OutAmsTestBase.java</include>
                     </includes>
                 </configuration>
             </plugin>
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java
index cd5533949..827fd8733 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java
@@ -48,12 +48,12 @@ import static io.javalin.apibuilder.ApiBuilder.path;
 import static io.javalin.apibuilder.ApiBuilder.post;
 import static io.javalin.apibuilder.ApiBuilder.put;
 
-
 public class AmsRestServer {
   public static final Logger LOG = LoggerFactory.getLogger("AmsRestServer");
+  private static Javalin app;
 
   public static void startRestServer(Integer port) {
-    Javalin app = Javalin.create(config -> {
+    app = Javalin.create(config -> {
       config.addStaticFiles(staticFiles -> {
         staticFiles.hostedPath = "/";
         // change to host files on a subpath, like '/assets'
@@ -82,7 +82,8 @@ public class AmsRestServer {
 
       config.sessionHandler(() -> new SessionHandler());
       config.enableCorsForAllOrigins();
-    }).start(port);
+    });
+    app.start(port);
     LOG.info("Javalin Rest server start at {}!!!", port);
 
     // before
@@ -94,7 +95,7 @@ public class AmsRestServer {
       } else if (needLoginCheck(uriPath)) {
         if (null == ctx.sessionAttribute("user")) {
           LOG.info("session info: {}", ctx.sessionAttributeMap() == null ? null : JSONObject.toJSONString(
-                  ctx.sessionAttributeMap()));
+              ctx.sessionAttributeMap()));
           throw new ForbiddenException();
         }
       }
@@ -214,11 +215,17 @@ public class AmsRestServer {
       ctx.json(new ErrorResponse(HttpCode.NOT_FOUND, "page not found!", ""));
     });
 
-    app.error(HttpCode.INTERNAL_SERVER_ERROR.getStatus(),ctx -> {
+    app.error(HttpCode.INTERNAL_SERVER_ERROR.getStatus(), ctx -> {
       ctx.json(new ErrorResponse(HttpCode.INTERNAL_SERVER_ERROR, "internal error!", ""));
     });
   }
 
+  public static void stopRestServer() {
+    if (app != null) {
+      app.stop();
+    }
+  }
+
   private static final String[] urlWhiteList = {
     "/ams/v1/login",
     "/",
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java
index cf852ec80..90f642238 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java
@@ -27,6 +27,9 @@ import com.netease.arctic.ams.api.InvalidObjectException;
 import com.netease.arctic.ams.api.MetaException;
 import com.netease.arctic.ams.api.NoSuchObjectException;
 import com.netease.arctic.ams.api.OptimizeManager;
+import com.netease.arctic.ams.api.client.AmsServerInfo;
+import com.netease.arctic.ams.api.client.ZookeeperService;
+import com.netease.arctic.ams.api.properties.AmsHAProperties;
 import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
 import com.netease.arctic.ams.server.config.ArcticMetaStoreConf;
 import com.netease.arctic.ams.server.config.ConfigFileProperties;
@@ -42,9 +45,11 @@ import com.netease.arctic.ams.server.service.impl.DerbyService;
 import com.netease.arctic.ams.server.service.impl.FileInfoCacheService;
 import com.netease.arctic.ams.server.service.impl.OptimizeExecuteService;
 import com.netease.arctic.ams.server.service.impl.RuntimeDataExpireService;
+import com.netease.arctic.ams.server.utils.AmsUtils;
 import com.netease.arctic.ams.server.utils.SecurityUtils;
 import com.netease.arctic.ams.server.utils.ThreadPool;
 import com.netease.arctic.ams.server.utils.YamlUtils;
+import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.thrift.TMultiplexedProcessor;
 import org.apache.thrift.protocol.TBinaryProtocol;
@@ -58,9 +63,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.net.InetAddress;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.List;
@@ -76,12 +83,31 @@ public class ArcticMetaStore {
 
   public static Configuration conf;
   private static JSONObject yamlConfig;
+  private static TServer server;
+  private static final List<Thread> residentThreads = new ArrayList<>();
+  private static HighAvailabilityServices haService = null;
+  private static final AtomicBoolean isLeader = new AtomicBoolean(false);
+  private static final int checkLeaderInterval = 2000;
 
   public static void main(String[] args) throws Throwable {
+    tryStartServer();
+  }
+
+  public static void tryStartServer() throws Throwable {
     try {
       String configPath = System.getenv(ArcticMetaStoreConf.ARCTIC_HOME.key()) + "/conf/config.yaml";
       yamlConfig = YamlUtils.load(configPath);
-      startMetaStore(initSystemConfig());
+      Configuration conf = initSystemConfig();
+      ArcticMetaStore.conf = conf;
+      if (conf.getBoolean(ArcticMetaStoreConf.HA_ENABLE)) {
+        String zkAddress = conf.getString(ArcticMetaStoreConf.ZOOKEEPER_SERVER);
+        String cluster = conf.getString(ArcticMetaStoreConf.CLUSTER_NAME);
+        haService = HighAvailabilityServices.getInstance(zkAddress, cluster);
+        haService.addListener(genHAListener(zkAddress, cluster));
+        haService.leaderLatch();
+      } else {
+        startMetaStore(conf);
+      }
     } catch (Throwable t) {
       LOG.error("MetaStore Thrift Server threw an exception...", t);
       throw t;
@@ -90,7 +116,6 @@ public class ArcticMetaStore {
 
   public static void startMetaStore(Configuration conf) throws Throwable {
     try {
-      ArcticMetaStore.conf = conf;
       long maxMessageSize = conf.getLong(ArcticMetaStoreConf.SERVER_MAX_MESSAGE_SIZE);
       int minWorkerThreads = conf.getInteger(ArcticMetaStoreConf.SERVER_MIN_THREADS);
       int maxWorkerThreads = conf.getInteger(ArcticMetaStoreConf.SERVER_MAX_THREADS);
@@ -132,7 +157,7 @@ public class ArcticMetaStore {
           .protocolFactory(protocolFactory)
           .inputProtocolFactory(inputProtoFactory)
           .workerThreads(maxWorkerThreads);
-      TServer server = new TThreadedSelectorServer(args);
+      server = new TThreadedSelectorServer(args);
       LOG.info("Started the new meta server on port [" + port + "]...");
       LOG.info("Options.minWorkerThreads = " + minWorkerThreads);
       LOG.info("Options.maxWorkerThreads = " + maxWorkerThreads);
@@ -149,6 +174,9 @@ public class ArcticMetaStore {
       signalOtherThreadsToStart(server, metaStoreThreadsLock, startCondition, startedServing);
       syncAndExpiredFileInfoCache(server);
       startSyncDDl(server);
+      if (conf.getBoolean(ArcticMetaStoreConf.HA_ENABLE)) {
+        checkLeader();
+      }
       server.serve();
     } catch (Throwable t) {
       LOG.error("ams start error", t);
@@ -156,6 +184,19 @@ public class ArcticMetaStore {
     }
   }
 
+  public static void stopMetaStore() {
+    if (server != null && server.isServing()) {
+      server.stop();
+    }
+    residentThreads.forEach(Thread::interrupt);
+    ThreadPool.shutdown();
+  }
+
+  public static void failover() {
+    stopMetaStore();
+    AmsRestServer.stopRestServer();
+  }
+
   private static void startMetaStoreThreads(
       Configuration conf,
       final Lock startLock,
@@ -187,6 +228,7 @@ public class ArcticMetaStore {
 
     t.setName("Metastore threads starter thread");
     t.start();
+    residentThreads.add(t);
   }
 
   private static void signalOtherThreadsToStart(
@@ -213,6 +255,7 @@ public class ArcticMetaStore {
       }
     });
     t.start();
+    residentThreads.add(t);
   }
 
   private static void startOptimizeCheck(final long checkInterval) {
@@ -288,6 +331,7 @@ public class ArcticMetaStore {
       }
     });
     t.start();
+    residentThreads.add(t);
   }
 
   private static void startSyncDDl(final TServer server) {
@@ -315,6 +359,35 @@ public class ArcticMetaStore {
       }
     });
     t.start();
+    residentThreads.add(t);
+  }
+
+  private static void checkLeader() {
+    Thread t = new Thread(() -> {
+      while (true) {
+        try {
+          Thread.sleep(checkLeaderInterval);
+        } catch (InterruptedException e) {
+          LOG.warn("notLeader thread was interrupted: " + e.getMessage());
+        }
+        try {
+          if (haService != null && isLeader.get() &&
+              !haService.getMaster().equals(haService.getNodeInfo(
+                  conf.getString(ArcticMetaStoreConf.THRIFT_BIND_HOST),
+                  conf.getInteger(ArcticMetaStoreConf.THRIFT_BIND_PORT)))) {
+            LOG.info("there is not leader, the leader is " + JSONObject.toJSONString(haService.getMaster()));
+            failover();
+            isLeader.set(false);
+          }
+        } catch (Exception e) {
+          LOG.error("check leader error", e);
+          failover();
+          isLeader.set(false);
+        }
+      }
+    });
+    t.start();
+    residentThreads.add(t);
   }
 
   private static Configuration initSystemConfig() {
@@ -323,8 +396,16 @@ public class ArcticMetaStore {
     config.setString(ArcticMetaStoreConf.ARCTIC_HOME, System.getenv(ArcticMetaStoreConf.ARCTIC_HOME.key()));
     config.setInteger(
         ArcticMetaStoreConf.THRIFT_BIND_PORT, systemConfig.getInteger(ArcticMetaStoreConf.THRIFT_BIND_PORT.key()));
-    config.setString(
-        ArcticMetaStoreConf.THRIFT_BIND_HOST, systemConfig.getString(ArcticMetaStoreConf.THRIFT_BIND_HOST.key()));
+    if (!systemConfig.containsKey(ArcticMetaStoreConf.THRIFT_BIND_HOST_PREFIX.key())) {
+      throw new RuntimeException("configuration " + ArcticMetaStoreConf.THRIFT_BIND_HOST_PREFIX.key() + " must be set");
+    }
+    InetAddress inetAddress =
+        AmsUtils.getLocalHostExactAddress(systemConfig.getString(ArcticMetaStoreConf.THRIFT_BIND_HOST_PREFIX.key()));
+    if (inetAddress == null) {
+      throw new RuntimeException("can't find host address start with " +
+          systemConfig.getString(ArcticMetaStoreConf.THRIFT_BIND_HOST_PREFIX.key()));
+    }
+    config.setString(ArcticMetaStoreConf.THRIFT_BIND_HOST, inetAddress.getHostAddress());
     config.setInteger(
         ArcticMetaStoreConf.HTTP_SERVER_PORT, systemConfig.getInteger(ArcticMetaStoreConf.HTTP_SERVER_PORT.key()));
     config.setInteger(
@@ -352,6 +433,8 @@ public class ArcticMetaStore {
     config.setString(
         ArcticMetaStoreConf.MYBATIS_CONNECTION_DRIVER_CLASS_NAME,
         systemConfig.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_DRIVER_CLASS_NAME.key()));
+
+    //mysql config
     if (systemConfig.getString(ArcticMetaStoreConf.DB_TYPE.key()).equalsIgnoreCase("mysql")) {
       config.setString(
           ArcticMetaStoreConf.MYBATIS_CONNECTION_PASSWORD,
@@ -360,6 +443,19 @@ public class ArcticMetaStore {
           ArcticMetaStoreConf.MYBATIS_CONNECTION_USER_NAME,
           systemConfig.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_USER_NAME.key()));
     }
+
+    //HA config
+    if (systemConfig.containsKey(ArcticMetaStoreConf.HA_ENABLE.key()) &&
+        systemConfig.getBoolean(ArcticMetaStoreConf.HA_ENABLE.key())) {
+      config.setBoolean(ArcticMetaStoreConf.HA_ENABLE, true);
+      config.setString(
+          ArcticMetaStoreConf.CLUSTER_NAME,
+          systemConfig.getString(ArcticMetaStoreConf.CLUSTER_NAME.key()));
+      config.setString(
+          ArcticMetaStoreConf.ZOOKEEPER_SERVER,
+          systemConfig.getString(ArcticMetaStoreConf.ZOOKEEPER_SERVER.key()));
+    }
+
     //extension properties
     String extensionPro = yamlConfig.getString(ConfigFileProperties.SYSTEM_EXTENSION_CONFIG) == null ? "" :
         yamlConfig.getString(ConfigFileProperties.SYSTEM_EXTENSION_CONFIG);
@@ -507,4 +603,32 @@ public class ArcticMetaStore {
       }
     }
   }
+
+  private static LeaderLatchListener genHAListener(String zkServerAddress, String namespace) {
+    return new LeaderLatchListener() {
+      @Override
+      public void isLeader() {
+        LOG.info("i am leader");
+        try {
+          ZookeeperService zkService = ZookeeperService.getInstance(zkServerAddress);
+          String masterPath = AmsHAProperties.getMasterPath(namespace);
+          zkService.create(masterPath);
+          AmsServerInfo serverInfo = new AmsServerInfo();
+          serverInfo.setHost(conf.getString(ArcticMetaStoreConf.THRIFT_BIND_HOST));
+          serverInfo.setThriftBindPort(conf.getInteger(ArcticMetaStoreConf.THRIFT_BIND_PORT));
+          zkService.setData(masterPath, JSONObject.toJSONString(serverInfo));
+          isLeader.set(true);
+          startMetaStore(initSystemConfig());
+        } catch (Throwable throwable) {
+          failover();
+        }
+      }
+
+      @Override
+      public void notLeader() {
+        //do nothing
+        LOG.info("i am salver");
+      }
+    };
+  }
 }
\ No newline at end of file
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/HighAvailabilityServices.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/HighAvailabilityServices.java
new file mode 100644
index 000000000..ac24f10ea
--- /dev/null
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/HighAvailabilityServices.java
@@ -0,0 +1,91 @@
+/*
+ * 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.netease.arctic.ams.server;
+
+import com.alibaba.fastjson.JSONObject;
+import com.netease.arctic.ams.api.client.AmsServerInfo;
+import com.netease.arctic.ams.api.properties.AmsHAProperties;
+import com.netease.arctic.ams.server.utils.ZookeeperUtils;
+import org.apache.curator.framework.recipes.leader.LeaderLatch;
+import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HighAvailabilityServices {
+
+  public static final Logger LOG = LoggerFactory.getLogger(HighAvailabilityServices.class);
+
+  private final LeaderLatch leaderLatch;
+  private static volatile HighAvailabilityServices instance;
+  private static final List<LeaderLatchListener> listeners = new ArrayList<>();
+  private static ZookeeperUtils zkService;
+  private final String namespace;
+
+  private HighAvailabilityServices(String zkServerAddress, String namespace) {
+    this.namespace = namespace;
+    zkService = new ZookeeperUtils(zkServerAddress);
+    String lockPath = AmsHAProperties.getLeaderPath(namespace);
+    try {
+      zkService.create(lockPath);
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+    leaderLatch = new LeaderLatch(zkService.getZkClient(), lockPath);
+  }
+
+  public static HighAvailabilityServices getInstance(String zkServerAddress, String namespace) {
+    Preconditions.checkNotNull(namespace, "cluster name cannot be null");
+
+    if (instance == null) {
+      synchronized (HighAvailabilityServices.class) {
+        if (instance == null) {
+          instance = new HighAvailabilityServices(zkServerAddress, namespace);
+        }
+      }
+    }
+    return instance;
+  }
+
+  public void addListener(LeaderLatchListener listener) {
+    listeners.add(listener);
+  }
+
+  public void leaderLatch() throws Exception {
+    listeners.forEach(leaderLatch::addListener);
+    leaderLatch.start();
+    leaderLatch.await();
+  }
+
+  public AmsServerInfo getNodeInfo(String host, int port) {
+    AmsServerInfo amsServerInfo = new AmsServerInfo();
+    amsServerInfo.setHost(host);
+    amsServerInfo.setThriftBindPort(port);
+    return amsServerInfo;
+  }
+
+  public AmsServerInfo getMaster() throws Exception {
+    return JSONObject.parseObject(
+        zkService.getData(AmsHAProperties.getMasterPath(namespace)),
+        AmsServerInfo.class);
+  }
+}
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ArcticMetaStoreConf.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ArcticMetaStoreConf.java
index 8c111ec79..054f90e7b 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ArcticMetaStoreConf.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ArcticMetaStoreConf.java
@@ -19,6 +19,18 @@
 package com.netease.arctic.ams.server.config;
 
 public class ArcticMetaStoreConf {
+  public static final ConfigOption<String> CLUSTER_NAME =
+      ConfigOptions.key("arctic.ams.cluster.name")
+          .stringType()
+          .defaultValue("default")
+          .withDescription("arctic ams cluster name.");
+
+  public static final ConfigOption<Boolean> HA_ENABLE =
+      ConfigOptions.key("arctic.ams.ha.enable")
+          .booleanType()
+          .defaultValue(false)
+          .withDescription("is arctic ams running high available.");
+
   public static final ConfigOption<Long> SERVER_MAX_MESSAGE_SIZE =
       ConfigOptions.key("arctic.ams.server.max.message.size")
           .longType()
@@ -91,6 +103,16 @@ public class ArcticMetaStoreConf {
           .stringType()
           .defaultValue("")
           .withDescription("Bind host on which to run the ams thrift service.");
+  public static final ConfigOption<String> THRIFT_BIND_HOST_PREFIX =
+      ConfigOptions.key("arctic.ams.server-host.prefix")
+          .stringType()
+          .defaultValue("")
+          .withDescription("Bind host on which to run the ams thrift service.");
+  public static final ConfigOption<String> ZOOKEEPER_SERVER =
+      ConfigOptions.key("arctic.ams.zookeeper.server")
+          .stringType()
+          .defaultValue("")
+          .withDescription("zookeeper server uri.");
   public static final ConfigOption<Boolean> USE_THRIFT_COMPACT_PROTOCOL =
       ConfigOptions.key("arctic.ams.thrift.compact.protocol.enabled")
           .booleanType()
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java
index 581599d4f..6bfcb4447 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java
@@ -29,6 +29,11 @@ import org.apache.thrift.TException;
 
 public class OptimizeManagerHandler implements OptimizeManager.Iface {
 
+  @Override
+  public void ping() throws TException {
+
+  }
+
   @Override
   public OptimizeTask pollTask(int queueId, JobId jobId, String attemptId, long waitTime)
       throws TException {
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/SnapInfoCacheMapper.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/SnapInfoCacheMapper.java
index a94ffe8ea..7515dde5c 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/SnapInfoCacheMapper.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/SnapInfoCacheMapper.java
@@ -30,6 +30,7 @@ import org.apache.ibatis.annotations.Result;
 import org.apache.ibatis.annotations.Results;
 import org.apache.ibatis.annotations.Select;
 
+import java.sql.Timestamp;
 import java.util.List;
 
 public interface SnapInfoCacheMapper {
@@ -57,7 +58,8 @@ public interface SnapInfoCacheMapper {
       ".server.mybatis.TableIdentifier2StringConverter} and inner_table = #{type} and snapshot_id not in (select " +
       "add_snapshot_id from file_info_cache where delete_snapshot_id is null and table_identifier=#{tableIdentifier, " +
       "typeHandler=com.netease.arctic.ams.server.mybatis.TableIdentifier2StringConverter} and inner_table = #{type})")
-  void expireCache(@Param("expiredTime") long expiredTime, @Param("tableIdentifier") TableIdentifier tableIdentifier,
+  void expireCache(
+      @Param("expiredTime") long expiredTime, @Param("tableIdentifier") TableIdentifier tableIdentifier,
       @Param("type") String tableType);
 
   @Delete("delete from " + TABLE_NAME + " where table_identifier = #{tableIdentifier, typeHandler=com.netease.arctic" +
@@ -66,13 +68,16 @@ public interface SnapInfoCacheMapper {
 
   @Delete("delete from " + TABLE_NAME + " where table_identifier = #{tableIdentifier, typeHandler=com.netease.arctic" +
       ".ams.server.mybatis.TableIdentifier2StringConverter} and inner_table = #{innerTable}")
-  void deleteInnerTableCache(@Param("tableIdentifier") TableIdentifier tableIdentifier,
+  void deleteInnerTableCache(
+      @Param("tableIdentifier") TableIdentifier tableIdentifier,
       @Param("innerTable") String innerTable);
 
   @Select("select max(commit_time) from " + TABLE_NAME + " where table_identifier = " +
       "#{tableIdentifier, typeHandler=com.netease.arctic.ams.server.mybatis.TableIdentifier2StringConverter} and " +
       "inner_table = #{type}")
-  Long getCachedMaxTime(@Param("tableIdentifier") TableIdentifier tableIdentifier, @Param("type") String tableType);
+  Timestamp getCachedMaxTime(
+      @Param("tableIdentifier") TableIdentifier tableIdentifier,
+      @Param("type") String tableType);
 
   @Select(
       "select snapshot_id, file_count as cnt, file_size as size, commit_time from " + TABLE_NAME + " where " +
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
index c4317b175..6d2f6235e 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
@@ -124,7 +124,8 @@ public class FileInfoCacheService extends IJDBCService {
   public Long getCachedMaxTime(TableIdentifier identifier, String innerTable) {
     try (SqlSession sqlSession = getSqlSession(true)) {
       SnapInfoCacheMapper snapInfoCacheMapper = getMapper(sqlSession, SnapInfoCacheMapper.class);
-      return snapInfoCacheMapper.getCachedMaxTime(identifier, innerTable);
+      Timestamp maxTime =  snapInfoCacheMapper.getCachedMaxTime(identifier, innerTable);
+      return maxTime == null ? 0 : maxTime.getTime();
     }
   }
 
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
index 09b461663..ab4b06535 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
@@ -53,12 +53,24 @@ public class OptimizeExecuteService {
     systemInfo.put(
         ArcticMetaStoreConf.ARCTIC_HOME.key(),
         ArcticMetaStore.conf.getString(ArcticMetaStoreConf.ARCTIC_HOME));
-    systemInfo.put(
-            ArcticMetaStoreConf.THRIFT_BIND_HOST.key(),
-            ArcticMetaStore.conf.getString(ArcticMetaStoreConf.THRIFT_BIND_HOST));
-    systemInfo.put(
-            ArcticMetaStoreConf.THRIFT_BIND_PORT.key(),
-            String.valueOf(ArcticMetaStore.conf.getInteger(ArcticMetaStoreConf.THRIFT_BIND_PORT)));
+    if (ArcticMetaStore.conf.getBoolean(ArcticMetaStoreConf.HA_ENABLE, false)) {
+      systemInfo.put(
+          ArcticMetaStoreConf.HA_ENABLE.key(),
+          ArcticMetaStore.conf.getBoolean(ArcticMetaStoreConf.HA_ENABLE));
+      systemInfo.put(
+          ArcticMetaStoreConf.CLUSTER_NAME.key(),
+          ArcticMetaStore.conf.getString(ArcticMetaStoreConf.CLUSTER_NAME));
+      systemInfo.put(
+          ArcticMetaStoreConf.ZOOKEEPER_SERVER.key(),
+          String.valueOf(ArcticMetaStore.conf.getString(ArcticMetaStoreConf.ZOOKEEPER_SERVER)));
+    } else {
+      systemInfo.put(
+          ArcticMetaStoreConf.THRIFT_BIND_HOST.key(),
+          ArcticMetaStore.conf.getString(ArcticMetaStoreConf.THRIFT_BIND_HOST));
+      systemInfo.put(
+          ArcticMetaStoreConf.THRIFT_BIND_PORT.key(),
+          String.valueOf(ArcticMetaStore.conf.getInteger(ArcticMetaStoreConf.THRIFT_BIND_PORT)));
+    }
     properties.put(AMS_SYSTEM_INFO, JSONObject.toJSONString(systemInfo));
     properties.put(CONTAINER_INFO, JSONObject.toJSONString(container));
     properties.put(OPTIMIZER_GROUP_INFO, JSONObject.toJSONString(groupInfo));
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/AmsUtils.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/AmsUtils.java
index 97af867f9..2563c55af 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/AmsUtils.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/AmsUtils.java
@@ -26,6 +26,9 @@ import com.netease.arctic.ams.server.model.OptimizeHistory;
 import com.netease.arctic.ams.server.model.TransactionsOfTable;
 
 import java.io.File;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.Enumeration;
 
 public class AmsUtils {
   public static TableIdentifier toTableIdentifier(com.netease.arctic.table.TableIdentifier tableIdentifier) {
@@ -101,4 +104,41 @@ public class AmsUtils {
     return path == null ? null : new File(path).getName();
   }
 
+  public static InetAddress getLocalHostExactAddress(String prefix) {
+    try {
+      InetAddress candidateAddress = null;
+      Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
+      while (networkInterfaces.hasMoreElements()) {
+        NetworkInterface iface = networkInterfaces.nextElement();
+        for (Enumeration<InetAddress> inetAddrs = iface.getInetAddresses(); inetAddrs.hasMoreElements(); ) {
+          InetAddress inetAddr = inetAddrs.nextElement();
+          if (!inetAddr.isLoopbackAddress()) {
+            if (inetAddr.isSiteLocalAddress() && checkHostAddress(inetAddr, prefix)) {
+              return inetAddr;
+            }
+            if (candidateAddress == null) {
+              candidateAddress = inetAddr;
+            }
+          }
+        }
+      }
+      if (candidateAddress != null) {
+        if (checkHostAddress(candidateAddress, prefix)) {
+          return candidateAddress;
+        }
+      } else {
+        if (checkHostAddress(InetAddress.getLocalHost(), prefix)) {
+          return InetAddress.getLocalHost();
+        }
+      }
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+    return null;
+  }
+
+  private static boolean checkHostAddress(InetAddress address, String prefix) {
+    return address != null && address.getHostAddress().startsWith(prefix);
+  }
+
 }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/JDBCSqlSessionFactoryProvider.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/JDBCSqlSessionFactoryProvider.java
index 417265ae8..c89975149 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/JDBCSqlSessionFactoryProvider.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/JDBCSqlSessionFactoryProvider.java
@@ -51,6 +51,8 @@ import org.apache.ibatis.session.SqlSessionFactory;
 import org.apache.ibatis.transaction.TransactionFactory;
 import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory;
 
+import java.time.Duration;
+
 public class JDBCSqlSessionFactoryProvider {
 
   private static volatile SqlSessionFactory sqlSessionFactory;
@@ -63,13 +65,13 @@ public class JDBCSqlSessionFactoryProvider {
           if (ArcticMetaStore.conf.getString(ArcticMetaStoreConf.DB_TYPE).equals("derby")) {
             dataSource.setUrl(ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_URL));
             dataSource.setDriverClassName(
-                    ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_DRIVER_CLASS_NAME));
+                ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_DRIVER_CLASS_NAME));
           } else {
             dataSource.setUsername(ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_USER_NAME));
             dataSource.setPassword(ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_PASSWORD));
             dataSource.setUrl(ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_URL));
             dataSource.setDriverClassName(
-                    ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_DRIVER_CLASS_NAME));
+                ArcticMetaStore.conf.getString(ArcticMetaStoreConf.MYBATIS_CONNECTION_DRIVER_CLASS_NAME));
           }
           dataSource.setDefaultAutoCommit(true);
           dataSource.setMaxTotal(20);
@@ -79,8 +81,7 @@ public class JDBCSqlSessionFactoryProvider {
           dataSource.setLogAbandoned(true);
           dataSource.setRemoveAbandonedOnBorrow(true);
           dataSource.setRemoveAbandonedTimeout(60);
-          dataSource.setTimeBetweenEvictionRunsMillis(
-              BaseObjectPoolConfig.DEFAULT_TIME_BETWEEN_EVICTION_RUNS.toMillis());
+          dataSource.setTimeBetweenEvictionRunsMillis(Duration.ofMillis(10 * 60 * 1000L).toMillis());
           dataSource.setTestOnBorrow(BaseObjectPoolConfig.DEFAULT_TEST_ON_BORROW);
           dataSource.setTestWhileIdle(BaseObjectPoolConfig.DEFAULT_TEST_WHILE_IDLE);
           dataSource.setMinEvictableIdleTimeMillis(1000);
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ThreadPool.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ThreadPool.java
index 0378ae449..d89834638 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ThreadPool.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ThreadPool.java
@@ -146,6 +146,7 @@ public class ThreadPool {
       commitPool.shutdown();
       expirePool.shutdown();
       orphanPool.shutdown();
+      optimizerMonitorPool.shutdown();
       syncFileInfoCachePool.shutdown();
       self = null;
     }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ZookeeperUtils.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ZookeeperUtils.java
new file mode 100644
index 000000000..e6a8b761f
--- /dev/null
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ZookeeperUtils.java
@@ -0,0 +1,91 @@
+/*
+ * 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.netease.arctic.ams.server.utils;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.data.Stat;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Provides ZooKeeper clients and operations.
+ */
+public class ZookeeperUtils {
+  private final CuratorFramework zkClient;
+  private static volatile ZookeeperUtils instance;
+
+  public ZookeeperUtils(String zkServerAddress) {
+    ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3, 5000);
+    this.zkClient = CuratorFrameworkFactory.builder()
+        .connectString(zkServerAddress)
+        .sessionTimeoutMs(5000)
+        .connectionTimeoutMs(5000)
+        .retryPolicy(retryPolicy)
+        .build();
+    zkClient.start();
+  }
+
+  public static ZookeeperUtils getInstance(String zkServerAddress) {
+    if (instance == null) {
+      synchronized (ZookeeperUtils.class) {
+        if (instance == null) {
+          instance = new ZookeeperUtils(zkServerAddress);
+        }
+      }
+    }
+    return instance;
+  }
+
+  public CuratorFramework getZkClient() {
+    return this.zkClient;
+  }
+
+  public boolean exist(String path) throws Exception {
+    Stat stat = zkClient.checkExists().forPath(path);
+    return stat != null;
+  }
+
+  public void create(String path) throws Exception {
+    StringBuilder tmpPath = new StringBuilder();
+    for (String p : path.split("/")) {
+      if (!p.isEmpty()) {
+        tmpPath.append("/");
+        tmpPath.append(p);
+        if (!exist(tmpPath.toString())) {
+          zkClient.create().withMode(CreateMode.PERSISTENT).forPath(tmpPath.toString());
+        }
+      }
+    }
+  }
+
+  public void setData(String path, String data) throws Exception {
+    zkClient.setData().forPath(path, data.getBytes(StandardCharsets.UTF_8));
+  }
+
+  public String getData(String path) throws Exception {
+    return new String(zkClient.getData().forPath(path), StandardCharsets.UTF_8);
+  }
+
+  public void delete(String path) throws Exception {
+    zkClient.delete().forPath(path);
+  }
+}
diff --git a/core/src/main/java/com/netease/arctic/PooledAmsClient.java b/core/src/main/java/com/netease/arctic/PooledAmsClient.java
index 03fb6fd77..16065cefb 100644
--- a/core/src/main/java/com/netease/arctic/PooledAmsClient.java
+++ b/core/src/main/java/com/netease/arctic/PooledAmsClient.java
@@ -23,6 +23,7 @@ import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.TableCommitMeta;
 import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.ams.api.TableMeta;
+import com.netease.arctic.ams.api.client.AmsClientPools;
 import org.apache.thrift.TException;
 
 import java.util.List;
diff --git a/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java b/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
index b3b4a6226..513949237 100644
--- a/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
+++ b/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
@@ -19,11 +19,11 @@
 package com.netease.arctic.catalog;
 
 import com.netease.arctic.AmsClient;
-import com.netease.arctic.AmsClientPools;
 import com.netease.arctic.PooledAmsClient;
 import com.netease.arctic.ams.api.ArcticTableMetastore;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.NoSuchObjectException;
+import com.netease.arctic.ams.api.client.AmsClientPools;
 import com.netease.arctic.ams.api.client.ArcticThriftUrl;
 import org.apache.iceberg.common.DynConstructors;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
diff --git a/dist/src/main/arctic-bin/bin/ams.sh b/dist/src/main/arctic-bin/bin/ams.sh
index 139da4728..bbe7bf38d 100644
--- a/dist/src/main/arctic-bin/bin/ams.sh
+++ b/dist/src/main/arctic-bin/bin/ams.sh
@@ -30,7 +30,6 @@ LIB_PATH=$ARCTIC_HOME/lib
 WORKDIR=$ARCTIC_HOME
 LOG_DIR=${WORKDIR}/logs
 STDERR_LOG=${WORKDIR}/logs/app.log.err
-STDOUT_LOG=${WORKDIR}/logs/app.log
 PID=${WORKDIR}/run/app.pid
 
 if [ ! -d "$LOG_DIR" ]; then
@@ -49,10 +48,6 @@ if [ ! -f $STDERR_LOG ];then
     touch $STDERR_LOG
 fi
 
-if [ ! -f $STDOUT_LOG ];then
-    touch $STDOUT_LOG
-fi
-
 if [ -z "$JAVA_OPTS" ]; then
     JAVA_OPTS="-Xms512m -Xmx512m -verbose:gc -XX:+PrintGCDetails"
 fi
@@ -71,7 +66,7 @@ function status(){
 }
 
 function start() {
-  nohup ${CMDS} </dev/null >>${STDOUT_LOG} 2>>${STDERR_LOG} &
+  nohup ${CMDS} </dev/null 2>>${STDERR_LOG} &
     if [ $? -ne 0 ]; then
         echo "start failed."
     fi
diff --git a/dist/src/main/arctic-bin/conf/config.yaml b/dist/src/main/arctic-bin/conf/config.yaml
index add125ad0..7531b6aef 100644
--- a/dist/src/main/arctic-bin/conf/config.yaml
+++ b/dist/src/main/arctic-bin/conf/config.yaml
@@ -1,5 +1,5 @@
 ams:
-  arctic.ams.server-host: 127.0.0.1
+  arctic.ams.server-host.prefix: "127." #To facilitate batch deployment can config server host prefix.Must be enclosed in double quotes
   arctic.ams.thrift.port: 1260
   arctic.ams.http.port: 1630
   arctic.ams.optimize.check.thread.pool-size: 10
@@ -18,6 +18,11 @@ ams:
   #arctic.ams.mybatis.ConnectionPassword: {password}
   #arctic.ams.database.type: mysql
 
+  #HA config
+  #arctic.ams.ha.enable: true
+  #arctic.ams.cluster.name: default
+  #arctic.ams.zookeeper.server: 127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183
+
 # extension properties for like system
 extension_properties:
 #test.properties: test
diff --git a/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOutputFileFactory.java b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOutputFileFactory.java
index 3325f203f..806c56fb5 100644
--- a/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOutputFileFactory.java
+++ b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOutputFileFactory.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.hive.io.writer;
 
+import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.writer.OutputFileFactory;
 import com.netease.arctic.io.writer.TaskWriterKey;
@@ -36,7 +37,7 @@ import java.util.concurrent.atomic.AtomicLong;
  *    ${table_location}
  *            -| change
  *            -| base
- *            -| hive_data
+ *            -| hive
  *                 -| ${partition_name1}
  *                 -| ${partition_name2}
  *                            -| txid=${txid}
@@ -45,13 +46,13 @@ import java.util.concurrent.atomic.AtomicLong;
  *    ${table_location}
  *            -| change
  *            -| base
- *            -| hive_data
+ *            -| hive
  *                  -| txid=${txid}
  *
  * For UnKeyed adapt hive with partitions the dir construct is :
  *    ${table_location}
  *            -| base
- *            -| hive_data
+ *            -| hive
  *                 -| ${partition_name1}
  *                 -| ${partition_name2}
  *                            -| ${timestamp_uuid}
@@ -59,7 +60,7 @@ import java.util.concurrent.atomic.AtomicLong;
  * For Keyed adapt hive without partitions the dir construct is :
  *    ${table_location}
  *            -| base
- *            -| hive_data
+ *            -| hive
  *                  -| ${timestamp_uuid}
  */
 public class AdaptHiveOutputFileFactory implements OutputFileFactory {
@@ -73,9 +74,7 @@ public class AdaptHiveOutputFileFactory implements OutputFileFactory {
   private final long taskId;
   private final Long transactionId;
 
-  private final String unKeyedTmpDir = System.currentTimeMillis() + "_" + UUID.randomUUID();
-
-  private final String keyedTmpDir;
+  private final String unKeyedTmpDir = HiveTableUtil.getRandomSubDir();
 
   private final String unKeyedTableNameUUID = UUID.randomUUID().toString();
 
@@ -98,7 +97,6 @@ public class AdaptHiveOutputFileFactory implements OutputFileFactory {
     this.partitionId = partitionId;
     this.taskId = taskId;
     this.transactionId = transactionId;
-    keyedTmpDir = "txid=" + transactionId;
   }
 
 
@@ -115,21 +113,13 @@ public class AdaptHiveOutputFileFactory implements OutputFileFactory {
   }
 
   private String fileLocation(StructLike partitionData, String fileName, TaskWriterKey key) {
-    if (partitionSpec.isUnpartitioned()) {
-      if (key.getTreeNode() == null) {
-        return String.format("%s/%s/%s", baseLocation, unKeyedTmpDir, fileName);
-      } else {
-        return String.format("%s/%s/%s", baseLocation, keyedTmpDir, fileName);
-      }
+    String dir;
+    if (key.getTreeNode() == null) {
+      dir = HiveTableUtil.newUnKeyedHiveDataLocation(baseLocation, partitionSpec, partitionData, unKeyedTmpDir);
     } else {
-      if (key.getTreeNode() == null) {
-        return String.format("%s/%s/%s/%s", baseLocation, partitionSpec.partitionToPath(partitionData),
-            unKeyedTmpDir, fileName);
-      } else {
-        return String.format("%s/%s/%s/%s", baseLocation, partitionSpec.partitionToPath(partitionData),
-            keyedTmpDir, fileName);
-      }
+      dir = HiveTableUtil.newKeyedHiveDataLocation(baseLocation, partitionSpec, partitionData, transactionId);
     }
+    return String.format("%s/%s", dir, fileName);
   }
 
   public EncryptedOutputFile newOutputFile(TaskWriterKey key) {
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java b/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
index 16bfefeaa..d0e05dee7 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
@@ -463,7 +463,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   private void generateUnpartitionTableLocation() {
     if (this.addFiles.isEmpty()) {
-      unpartitionTableLocation = createEmptyLocationForHive();
+      unpartitionTableLocation = createUnpartitionEmptyLocationForHive();
     } else {
       unpartitionTableLocation = FileUtil.getFileDir(this.addFiles.get(0).path().toString());
     }
@@ -486,13 +486,14 @@ public class OverwriteHiveFiles implements OverwriteFiles {
     }
   }
 
-  private String createEmptyLocationForHive() {
+  private String createUnpartitionEmptyLocationForHive() {
     // create a new empty location for hive
     String newLocation;
     if (txId > 0) {
-      newLocation = table.hiveLocation() + "/txId=" + txId;
+      newLocation = HiveTableUtil.newKeyedHiveDataLocation(table.hiveLocation(), table.spec(), null, txId);
     } else {
-      newLocation = table.hiveLocation() + "/ts_" + System.currentTimeMillis();
+      newLocation = HiveTableUtil.newUnKeyedHiveDataLocation(table.hiveLocation(), table.spec(), null,
+          HiveTableUtil.getRandomSubDir());
     }
     OutputFile file = table.io().newOutputFile(newLocation + "/.keep");
     try {
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
index e71251aee..a8da6d17e 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
@@ -29,6 +29,7 @@ import org.apache.iceberg.DataFile;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
@@ -36,6 +37,7 @@ import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
 public class HiveTableUtil {
 
@@ -89,6 +91,29 @@ public class HiveTableUtil {
     return tableLocation + "/hive";
   }
 
+  public static String newKeyedHiveDataLocation(String hiveLocation, PartitionSpec partitionSpec,
+                                                StructLike partitionData, Long transactionId) {
+    if (partitionSpec.isUnpartitioned()) {
+      return String.format("%s/%s", hiveLocation, "txid=" + transactionId);
+    } else {
+      return String.format("%s/%s/%s", hiveLocation, partitionSpec.partitionToPath(partitionData),
+          "txid=" + transactionId);
+    }
+  }
+
+  public static String newUnKeyedHiveDataLocation(String hiveLocation, PartitionSpec partitionSpec,
+                                                  StructLike partitionData, String subDir) {
+    if (partitionSpec.isUnpartitioned()) {
+      return String.format("%s/%s", hiveLocation, subDir);
+    } else {
+      return String.format("%s/%s/%s", hiveLocation, partitionSpec.partitionToPath(partitionData), subDir);
+    }
+  }
+
+  public static String getRandomSubDir() {
+    return System.currentTimeMillis() + "_" + UUID.randomUUID();
+  }
+
   public static StorageDescriptor storageDescriptor(
       Schema schema, PartitionSpec partitionSpec, String location,
       FileFormat format) {
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java b/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
index feee56c68..8b2808692 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
@@ -88,9 +88,15 @@ public class FlinkOptimizer implements StatefulOptimizer {
     String jarPath = " " + arcticHome + "/plugin/optimize/OptimizeJob.jar ";
     String entryClass = this.getClass().getName();
     cmd += " -c " + entryClass + jarPath;
-    String amsUrl =
-        "thrift://" + systemInfo.getString(OptimizerProperties.THRIFT_BIND_HOST).trim() + ":" +
-            systemInfo.getString(OptimizerProperties.THRIFT_BIND_PORT).trim();
+    String amsUrl;
+    if (systemInfo.containsKey(OptimizerProperties.HA_ENABLE) && systemInfo.getBoolean(OptimizerProperties.HA_ENABLE)) {
+      amsUrl = String.format("zookeeper://%s/%s", systemInfo.getString(OptimizerProperties.ZOOKEEPER_SERVER).trim(),
+          systemInfo.getString(OptimizerProperties.CLUSTER_NAME)).trim();
+    } else {
+      amsUrl =
+          "thrift://" + systemInfo.getString(OptimizerProperties.THRIFT_BIND_HOST).trim() + ":" +
+              systemInfo.getString(OptimizerProperties.THRIFT_BIND_PORT).trim();
+    }
     int parallelism = jobInfo.getInteger(OptimizerProperties.OPTIMIZER_JOB_PARALLELISM);
     long heartBeatInterval = groupProperties.containsKey(OptimizerProperties.OPTIMIZER_GROUP_HEART_BEAT_INTERVAL) ?
         groupProperties.getLong(OptimizerProperties.OPTIMIZER_GROUP_HEART_BEAT_INTERVAL) :
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java b/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java
index 644cb1b18..683404de0 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java
@@ -80,9 +80,15 @@ public class LocalOptimizer implements StatefulOptimizer {
     JSONObject groupProperties = groupInfo.getJSONObject(OptimizerProperties.OPTIMIZER_GROUP_PROPERTIES);
 
     //add compact execute config
-    String amsUrl =
-        "thrift://" + systemInfo.getString(OptimizerProperties.THRIFT_BIND_HOST).trim() + ":" +
-            systemInfo.getString(OptimizerProperties.THRIFT_BIND_PORT).trim();
+    String amsUrl;
+    if (systemInfo.containsKey(OptimizerProperties.HA_ENABLE) && systemInfo.getBoolean(OptimizerProperties.HA_ENABLE)) {
+      amsUrl = String.format("zookeeper://%s/%s", systemInfo.getString(OptimizerProperties.ZOOKEEPER_SERVER).trim(),
+          systemInfo.getString(OptimizerProperties.CLUSTER_NAME)).trim();
+    } else {
+      amsUrl =
+          "thrift://" + systemInfo.getString(OptimizerProperties.THRIFT_BIND_HOST).trim() + ":" +
+              systemInfo.getString(OptimizerProperties.THRIFT_BIND_PORT).trim();
+    }
     int parallelism = jobInfo.getInteger(OptimizerProperties.OPTIMIZER_JOB_PARALLELISM);
     long heartBeatInterval = groupProperties.containsKey(OptimizerProperties.OPTIMIZER_GROUP_HEART_BEAT_INTERVAL) ?
         groupProperties.getLong(OptimizerProperties.OPTIMIZER_GROUP_HEART_BEAT_INTERVAL) :
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskConsumer.java b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskConsumer.java
index 238d6443b..95e9e7fe7 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskConsumer.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskConsumer.java
@@ -23,9 +23,9 @@ import com.netease.arctic.ams.api.JobType;
 import com.netease.arctic.ams.api.NoSuchObjectException;
 import com.netease.arctic.ams.api.OptimizeManager;
 import com.netease.arctic.ams.api.OptimizeTask;
+import com.netease.arctic.ams.api.client.OptimizeManagerClientPools;
 import com.netease.arctic.optimizer.OptimizerConfig;
 import com.netease.arctic.optimizer.TaskWrapper;
-import com.netease.arctic.optimizer.util.OptimizeManagerClients;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -73,7 +73,7 @@ public class BaseTaskConsumer implements Serializable {
 
   private OptimizeTask pollTask(int attemptId, long timeout) throws TException {
     try {
-      OptimizeManager.Iface optimizeManager = OptimizeManagerClients.getClient(config.getAmsUrl());
+      OptimizeManager.Iface optimizeManager = OptimizeManagerClientPools.getClient(config.getAmsUrl());
       return optimizeManager.pollTask(config.getQueueId(), jobId, attemptId + "", timeout);
     } catch (NoSuchObjectException e) {
       return null;
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskReporter.java b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskReporter.java
index aafb8c03e..369413ae5 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskReporter.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseTaskReporter.java
@@ -20,8 +20,8 @@ package com.netease.arctic.optimizer.operator;
 
 import com.netease.arctic.ams.api.OptimizeManager;
 import com.netease.arctic.ams.api.OptimizeTaskStat;
+import com.netease.arctic.ams.api.client.OptimizeManagerClientPools;
 import com.netease.arctic.optimizer.OptimizerConfig;
-import com.netease.arctic.optimizer.util.OptimizeManagerClients;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -81,7 +81,7 @@ public class BaseTaskReporter implements Serializable {
       throws TException {
     LOG.info("start reporting result: {}", printOptimizeTaskStat(optimizeTaskStat));
     try {
-      OptimizeManager.Iface compactManager = OptimizeManagerClients.getClient(thriftUrl);
+      OptimizeManager.Iface compactManager = OptimizeManagerClientPools.getClient(thriftUrl);
       compactManager.reportOptimizeResult(optimizeTaskStat);
     } catch (Throwable t) {
       LOG.error("failed to sending result, task: {} ", optimizeTaskStat.getTaskId(), t);
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseToucher.java b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseToucher.java
index 56b6f2842..c38356964 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseToucher.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/BaseToucher.java
@@ -21,8 +21,8 @@ package com.netease.arctic.optimizer.operator;
 import com.google.common.collect.Maps;
 import com.netease.arctic.ams.api.OptimizeManager;
 import com.netease.arctic.ams.api.OptimizerStateReport;
+import com.netease.arctic.ams.api.client.OptimizeManagerClientPools;
 import com.netease.arctic.optimizer.OptimizerConfig;
-import com.netease.arctic.optimizer.util.OptimizeManagerClients;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -60,7 +60,7 @@ public class BaseToucher implements Serializable {
       if (state == null) {
         state = EMPTY_STATE;
       }
-      OptimizeManager.Iface client = OptimizeManagerClients.getClient(config.getAmsUrl());
+      OptimizeManager.Iface client = OptimizeManagerClientPools.getClient(config.getAmsUrl());
       OptimizerStateReport report = new OptimizerStateReport();
       report.optimizerId = Long.parseLong(config.getOptimizerId());
       report.optimizerState = state;
diff --git a/pom.xml b/pom.xml
index 16e2df92d..39582863c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -64,6 +64,10 @@
         <cglib.version>2.2.2</cglib.version>
         <mockito.version>2.23.4</mockito.version>
         <parquet.version>1.12.0</parquet.version>
+        <hive.version>2.3.7</hive.version>
+        <curator.version>5.2.1</curator.version>
+        <zookeeper.version>3.7.1</zookeeper.version>
+        <fastjson.version>1.2.75</fastjson.version>
     </properties>
 
     <dependencyManagement>


Test patch:
diff --git a/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockZookeeperServer.java b/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockZookeeperServer.java
new file mode 100644
index 000000000..d918ca666
--- /dev/null
+++ b/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockZookeeperServer.java
@@ -0,0 +1,76 @@
+/*
+ * 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.netease.arctic.ams.api;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.test.TestingServer;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+
+/**
+ * Provides mock zookeeper server.
+ */
+public class MockZookeeperServer {
+
+  private static TestingServer server;
+  private static CuratorFramework client;
+
+  public static CuratorFramework getClient() {
+    if (client == null) {
+      try {
+        init();
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    }
+    return client;
+  }
+
+  public static String getUri() {
+    return "127.0.0.1:2181";
+  }
+
+  public static void stopServer() throws IOException {
+    server.stop();
+    client.close();
+  }
+
+  private static void init() throws Exception {
+    server = new TestingServer(2181, true);
+    server.start();
+
+    client = CuratorFrameworkFactory.newClient("127.0.0.1",
+        new ExponentialBackoffRetry(1000, 3));
+    client.start();
+  }
+
+  @Test
+  public void testFoobar() throws Exception {
+    System.out.println("client: " + client);
+    getClient().create().forPath("/test", "test-data".getBytes());
+
+    byte[] data = client.getData().forPath("/test");
+    System.out.println("data: " + new String(data));
+  }
+}
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/OutAmsTestBase.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/OutAmsTestBase.java
new file mode 100644
index 000000000..499575771
--- /dev/null
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/OutAmsTestBase.java
@@ -0,0 +1,47 @@
+/*
+ * 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.netease.arctic.ams.server;
+
+import com.netease.arctic.ams.server.controller.LoginControllerTest;
+import com.netease.arctic.ams.server.controller.OptimizerControllerTest;
+import com.netease.arctic.ams.server.controller.TableControllerTest;
+import com.netease.arctic.ams.server.controller.TerminalControllerTest;
+import com.netease.arctic.ams.server.optimize.TestBaseOptimizePlan;
+import com.netease.arctic.ams.server.optimize.TestExpiredFileClean;
+import com.netease.arctic.ams.server.optimize.TestMajorOptimizeCommit;
+import com.netease.arctic.ams.server.optimize.TestMajorOptimizePlan;
+import com.netease.arctic.ams.server.optimize.TestMinorOptimizeCommit;
+import com.netease.arctic.ams.server.optimize.TestMinorOptimizePlan;
+import com.netease.arctic.ams.server.optimize.TestOrphanFileClean;
+import com.netease.arctic.ams.server.service.TestDDLTracerService;
+import org.junit.runner.RunWith;
+import org.junit.runners.Suite;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.modules.junit4.PowerMockRunnerDelegate;
+
+@RunWith(Suite.class)
+@Suite.SuiteClasses({TestHighAvailabilityServices.class})
+@PowerMockIgnore({"org.apache.logging.log4j.*", "javax.management.*", "org.apache.http.conn.ssl.*",
+                  "com.amazonaws.http.conn.ssl.*",
+                  "javax.net.ssl.*", "org.apache.hadoop.*", "javax.*", "com.sun.org.apache.*", "org.apache.xerces.*",
+                  "javax.xml.parsers.*"})
+public class OutAmsTestBase {
+
+}
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/TestHighAvailabilityServices.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/TestHighAvailabilityServices.java
new file mode 100644
index 000000000..1b6da8a7a
--- /dev/null
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/TestHighAvailabilityServices.java
@@ -0,0 +1,62 @@
+/*
+ * 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.netease.arctic.ams.server;
+
+import com.alibaba.fastjson.JSONObject;
+import com.netease.arctic.ams.api.MockZookeeperServer;
+import com.netease.arctic.ams.api.client.AmsServerInfo;
+import com.netease.arctic.ams.api.client.ArcticThriftUrl;
+import com.netease.arctic.ams.api.client.ZookeeperService;
+import com.netease.arctic.ams.api.properties.AmsHAProperties;
+import org.apache.curator.framework.CuratorFramework;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestHighAvailabilityServices {
+  CuratorFramework client = MockZookeeperServer.getClient();
+
+  @Test
+  public void testClient() throws Exception {
+    String testHost = "127.0.0.1";
+    int testPort = 1260;
+    String testCluster = "testCluster";
+    String testCatalog = "testCatalog";
+    int socketTimeout = 1111;
+    ZookeeperService zkService = ZookeeperService.getInstance(MockZookeeperServer.getUri());
+    String masterPath = AmsHAProperties.getMasterPath(testCluster);
+    zkService.create(masterPath);
+
+    AmsServerInfo serverInfo = new AmsServerInfo();
+    serverInfo.setHost(testHost);
+    serverInfo.setThriftBindPort(testPort);
+    zkService.setData(masterPath, JSONObject.toJSONString(serverInfo));
+
+    String zkUrl = String.format("zookeeper://%s/%s/%s?socketTimeout=%d", MockZookeeperServer.getUri(), testCluster,
+        testCatalog, socketTimeout);
+    String thriftUrl =
+        String.format("thrift://%s:%d/%s?socketTimeout=%d", testHost, testPort, testCatalog, socketTimeout);
+    ArcticThriftUrl arcticThriftUrl = ArcticThriftUrl.parse(zkUrl);
+    Assert.assertEquals(thriftUrl, arcticThriftUrl.url());
+    Assert.assertEquals(testCatalog.toLowerCase(), arcticThriftUrl.catalogName());
+    Assert.assertEquals(testHost, arcticThriftUrl.host());
+    Assert.assertEquals(testPort, arcticThriftUrl.port());
+    Assert.assertEquals(1111, arcticThriftUrl.socketTimeout());
+    Assert.assertEquals("thrift", arcticThriftUrl.schema());
+  }
+}
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
index 890100cfe..3f50c5834 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
@@ -19,7 +19,7 @@
 package com.netease.arctic.ams.server.controller;
 
 import com.alibaba.fastjson.JSONObject;
-import com.netease.arctic.AmsClientPools;
+import com.netease.arctic.ams.api.client.AmsClientPools;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.Constants;
 import com.netease.arctic.ams.api.DataFileInfo;
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TerminalControllerTest.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TerminalControllerTest.java
index ea79ab84b..c8e482d33 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TerminalControllerTest.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TerminalControllerTest.java
@@ -19,7 +19,7 @@
 package com.netease.arctic.ams.server.controller;
 
 import com.alibaba.fastjson.JSONObject;
-import com.netease.arctic.AmsClientPools;
+import com.netease.arctic.ams.api.client.AmsClientPools;
 import com.netease.arctic.CatalogMetaTestUtil;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.MockArcticMetastoreServer;
diff --git a/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java b/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
index 33f49857f..d1186f10f 100644
--- a/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
+++ b/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
@@ -19,7 +19,7 @@
 package com.netease.arctic.hive;
 
 import com.netease.arctic.hive.catalog.TestArcticHiveCatalog;
-import com.netease.arctic.hive.io.AdaptHiveWriterTest;
+import com.netease.arctic.hive.io.TestAdaptHiveWriter;
 import com.netease.arctic.hive.op.AutoSyncHiveTest;
 import com.netease.arctic.hive.op.TestHiveSchemaUpdate;
 import com.netease.arctic.hive.op.TestOverwriteFiles;
@@ -37,7 +37,7 @@ import org.junit.runners.Suite;
     TestOverwriteFiles.class,
     TestHiveSchemaUpdate.class,
     HiveMetaSynchronizerTest.class,
-    AdaptHiveWriterTest.class,
+    TestAdaptHiveWriter.class,
     AutoSyncHiveTest.class
 })
 public class ArcticHiveTestMain {
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java b/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveWriter.java
similarity index 99%
rename from hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
rename to hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveWriter.java
index 1cdf19ef7..3a8de3a3e 100644
--- a/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
+++ b/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveWriter.java
@@ -48,7 +48,7 @@ import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-public class AdaptHiveWriterTest extends HiveTableTestBase {
+public class TestAdaptHiveWriter extends HiveTableTestBase {
 
   @Test
   public void testWriteTypeFromOperateKind(){
diff --git a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
index 910b2cf01..097db7342 100644
--- a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
+++ b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
@@ -17,12 +17,12 @@ package com.netease.arctic.spark;/*
  */
 
 import com.google.common.collect.Sets;
-import com.netease.arctic.AmsClientPools;
 import com.netease.arctic.CatalogMetaTestUtil;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.MockArcticMetastoreServer;
 import com.netease.arctic.ams.api.NoSuchObjectException;
 import com.netease.arctic.ams.api.TableMeta;
+import com.netease.arctic.ams.api.client.AmsClientPools;
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.ChangeAction;
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
index dd9f6bd04..b3203540d 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
@@ -18,7 +18,7 @@
 
 package com.netease.arctic.spark;
 
-import com.netease.arctic.AmsClientPools;
+import com.netease.arctic.ams.api.client.AmsClientPools;
 import com.netease.arctic.CatalogMetaTestUtil;
 import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.MockArcticMetastoreServer;