support duration data type and sync latest client

connector/src/main/java/org.apache.flink/connector/nebula/catalog/AbstractNebulaCatalog.java

@@ -9,6 +9,7 @@
 import com.vesoft.nebula.client.graph.data.HostAddress;
 import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.meta.MetaClient;
+import java.net.UnknownHostException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -57,7 +58,12 @@ public AbstractNebulaCatalog(String catalogName, String defaultDatabase, String
     public void open() throws CatalogException {
         // test metaClient connection
         List<HostAddress> hostAndPorts = NebulaUtils.getHostAndPorts(address);
-        MetaClient metaClient = new MetaClient(hostAndPorts);
+        MetaClient metaClient = null;
+        try {
+            metaClient = new MetaClient(hostAndPorts);
+        } catch (UnknownHostException e) {
+            throw new IllegalArgumentException("address is illegal, ", e);
+        }
         try {
             metaClient.connect();
             metaClient.close();

connector/src/main/java/org.apache.flink/connector/nebula/catalog/NebulaCatalog.java

@@ -24,6 +24,7 @@
 import com.vesoft.nebula.meta.IdName;
 import com.vesoft.nebula.meta.Schema;
 import com.vesoft.nebula.meta.TagItem;
+import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -53,7 +54,7 @@ public class NebulaCatalog extends AbstractNebulaCatalog {
     private final MetaClient metaClient;
 
     public NebulaCatalog(String catalogName, String defaultDatabase, String username, String pwd,
-                         String address) {
+                         String address) throws UnknownHostException {
         super(catalogName, defaultDatabase, username, pwd, address);
         this.hostAndPorts = NebulaUtils.getHostAndPorts(address);
         this.metaClient = new MetaClient(hostAndPorts);

connector/src/main/java/org.apache.flink/connector/nebula/catalog/factory/NebulaCatalogFactory.java

@@ -13,6 +13,7 @@
 import static org.apache.flink.table.descriptors.CatalogDescriptorValidator.CATALOG_PROPERTY_VERSION;
 import static org.apache.flink.table.descriptors.CatalogDescriptorValidator.CATALOG_TYPE;
 
+import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -31,12 +32,16 @@ public class NebulaCatalogFactory implements CatalogFactory {
     @Override
     public Catalog createCatalog(String name, Map<String, String> properties) {
         final DescriptorProperties prop = getValidatedProperties(properties);
-        return new NebulaCatalog(
-                name,
-                prop.getString(CATALOG_DEFAULT_DATABASE),
-                prop.getString(CATALOG_NEBULA_USERNAME),
-                prop.getString(CATALOG_NEBULA_PASSWORD),
-                prop.getString(CATALOG_NEBULA_ADDRESS));
+        try {
+            return new NebulaCatalog(
+                    name,
+                    prop.getString(CATALOG_DEFAULT_DATABASE),
+                    prop.getString(CATALOG_NEBULA_USERNAME),
+                    prop.getString(CATALOG_NEBULA_PASSWORD),
+                    prop.getString(CATALOG_NEBULA_ADDRESS));
+        } catch (UnknownHostException e) {
+            throw new IllegalArgumentException("address is illegal,", e);
+        }
     }
 
     @Override

connector/src/main/java/org.apache.flink/connector/nebula/connection/NebulaMetaConnectionProvider.java

@@ -18,6 +18,7 @@
 import com.vesoft.nebula.meta.Schema;
 import com.vesoft.nebula.meta.SpaceItem;
 import java.io.Serializable;
+import java.net.UnknownHostException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -35,7 +36,8 @@ public NebulaMetaConnectionProvider(NebulaClientOptions nebulaClientOptions) {
         this.nebulaClientOptions = nebulaClientOptions;
     }
 
-    public MetaClient getMetaClient() throws TException, ClientServerIncompatibleException {
+    public MetaClient getMetaClient() throws TException, ClientServerIncompatibleException,
+            UnknownHostException {
         List<HostAddress> addresses = nebulaClientOptions.getMetaAddress();
         int timeout = nebulaClientOptions.getTimeout();
         int retry = nebulaClientOptions.getConnectRetry();

connector/src/main/java/org.apache.flink/connector/nebula/source/NebulaRowConverter.java

@@ -59,6 +59,9 @@ public Row convert(BaseTableRow row) throws UnsupportedEncodingException {
                 record.setField(pos, valueWrapper.asGeography());
                 continue;
             }
+            if (valueWrapper.isDuration()) {
+                record.setField(pos, valueWrapper.asDuration());
+            }
         }
         return record;
     }

connector/src/main/java/org.apache.flink/connector/nebula/utils/NebulaCatalogUtils.java

@@ -5,6 +5,7 @@
 
 package org.apache.flink.connector.nebula.utils;
 
+import java.net.UnknownHostException;
 import org.apache.flink.connector.nebula.catalog.NebulaCatalog;
 
 /**
@@ -17,7 +18,7 @@ public class NebulaCatalogUtils {
      */
     public static NebulaCatalog createNebulaCatalog(String catalogName, String defaultSpace,
                                                     String address, String username,
-                                                    String password) {
+                                                    String password) throws UnknownHostException {
         return new NebulaCatalog(catalogName, defaultSpace, username, password, address);
     }
 }