sync the lastest java client (#42)

vesoft-inc · Dec 30, 2021 · 0643db2 · 0643db2
1 parent e9ea2e2
commit 0643db2
Show file tree

Hide file tree

Showing 6 changed files with 28 additions and 10 deletions.
diff --git a/connector/src/main/java/org.apache.flink/connector/nebula/catalog/AbstractNebulaCatalog.java b/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();

diff --git a/connector/src/main/java/org.apache.flink/connector/nebula/catalog/NebulaCatalog.java b/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);

diff --git a/...src/main/java/org.apache.flink/connector/nebula/catalog/factory/NebulaCatalogFactory.java b/...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

diff --git a/.../main/java/org.apache.flink/connector/nebula/connection/NebulaMetaConnectionProvider.java b/.../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();

diff --git a/connector/src/main/java/org.apache.flink/connector/nebula/source/NebulaRowConverter.java b/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;
  }

diff --git a/connector/src/main/java/org.apache.flink/connector/nebula/utils/NebulaCatalogUtils.java b/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);
  }
 }