add version verification

client/src/main/java/com/vesoft/nebula/client/graph/exception/ClientServerIncompatibleException.java

@@ -0,0 +1,18 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+package com.vesoft.nebula.client.graph.exception;
+
+/**
+ *
+ */
+public class ClientServerIncompatibleException extends Exception {
+ public ClientServerIncompatibleException(String message) {
+ super("Current client is not compatible with the remote server, please check the "
+ + "version: " + message);
+ }
+}
+

client/src/main/java/com/vesoft/nebula/client/graph/net/ConnObjectPool.java

@@ -2,6 +2,7 @@
 
 import com.vesoft.nebula.client.graph.NebulaPoolConfig;
 import com.vesoft.nebula.client.graph.data.HostAddress;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 import org.apache.commons.pool2.BasePooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
@@ -18,7 +19,7 @@ public ConnObjectPool(LoadBalancer loadBalancer, NebulaPoolConfig config) {
  }
 
  @Override
- public SyncConnection create() throws IOErrorException {
+ public SyncConnection create() throws IOErrorException, ClientServerIncompatibleException {
  HostAddress address = loadBalancer.getAddress();
  if (address == null) {
  throw new IOErrorException(IOErrorException.E_ALL_BROKEN,
@@ -83,11 +84,11 @@ public void activateObject(PooledObject<SyncConnection> p) throws Exception {
  super.activateObject(p);
  }
 
- public boolean init() {
+ public boolean init() throws ClientServerIncompatibleException {
  return loadBalancer.isServersOK();
  }
 
- public void updateServerStatus() {
+ public void updateServerStatus() throws ClientServerIncompatibleException {
  loadBalancer.updateServersStatus();
  }
 }

client/src/main/java/com/vesoft/nebula/client/graph/net/Connection.java

@@ -2,6 +2,7 @@
 
 import com.vesoft.nebula.client.graph.data.HostAddress;
 import com.vesoft.nebula.client.graph.data.SSLParam;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 
 public abstract class Connection {
@@ -12,11 +13,12 @@ public HostAddress getServerAddress() {
  }
 
  public abstract void open(HostAddress address, int timeout, SSLParam sslParam)
- throws IOErrorException;
+ throws IOErrorException, ClientServerIncompatibleException;
 
- public abstract void open(HostAddress address, int timeout) throws IOErrorException;
+ public abstract void open(HostAddress address, int timeout) throws IOErrorException,
+ ClientServerIncompatibleException;
 
- public abstract void reopen() throws IOErrorException;
+ public abstract void reopen() throws IOErrorException, ClientServerIncompatibleException;
 
  public abstract void close();
 

client/src/main/java/com/vesoft/nebula/client/graph/net/LoadBalancer.java

@@ -1,13 +1,14 @@
 package com.vesoft.nebula.client.graph.net;
 
 import com.vesoft.nebula.client.graph.data.HostAddress;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 
 public interface LoadBalancer {
  HostAddress getAddress();
 
  void close();
 
- void updateServersStatus();
+ void updateServersStatus() throws ClientServerIncompatibleException;
 
- boolean isServersOK();
+ boolean isServersOK() throws ClientServerIncompatibleException;
 }

client/src/main/java/com/vesoft/nebula/client/graph/net/NebulaPool.java

@@ -9,6 +9,7 @@
 import com.vesoft.nebula.client.graph.NebulaPoolConfig;
 import com.vesoft.nebula.client.graph.data.HostAddress;
 import com.vesoft.nebula.client.graph.exception.AuthFailedException;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 import com.vesoft.nebula.client.graph.exception.InvalidConfigException;
 import com.vesoft.nebula.client.graph.exception.NotValidConnectionException;
@@ -79,7 +80,7 @@ private void checkConfig(NebulaPoolConfig config) {
  * @throws InvalidConfigException if config is illegal
  */
  public boolean init(List<HostAddress> addresses, NebulaPoolConfig config)
- throws UnknownHostException, InvalidConfigException {
+ throws UnknownHostException, InvalidConfigException, ClientServerIncompatibleException {
  checkInit();
  hasInit.set(true);
  checkConfig(config);
@@ -129,7 +130,8 @@ public void close() {
  * @throws AuthFailedException if authenticate failed
  */
  public Session getSession(String userName, String password, boolean reconnect)
- throws NotValidConnectionException, IOErrorException, AuthFailedException {
+ throws NotValidConnectionException, IOErrorException, AuthFailedException,
+ ClientServerIncompatibleException {
  checkNoInitAndClosed();
  SyncConnection connection = null;
  try {
@@ -177,7 +179,7 @@ public int getWaitersNum() {
  * Update the services' status when the connection is broken,
  * it is called by Session and NebulaPool
  */
- protected void updateServerStatus() {
+ protected void updateServerStatus() throws ClientServerIncompatibleException {
  checkNoInitAndClosed();
  if (objectPool.getFactory() instanceof ConnObjectPool) {
  ((ConnObjectPool)objectPool.getFactory()).updateServerStatus();

client/src/main/java/com/vesoft/nebula/client/graph/net/RoundRobinLoadBalancer.java

@@ -2,6 +2,7 @@
 
 import com.vesoft.nebula.client.graph.data.HostAddress;
 import com.vesoft.nebula.client.graph.data.SSLParam;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -59,7 +60,7 @@ public HostAddress getAddress() {
  return null;
  }
 
- public void updateServersStatus() {
+ public void updateServersStatus() throws ClientServerIncompatibleException {
  for (HostAddress addr : addresses) {
  if (ping(addr)) {
  serversStatus.put(addr, S_OK);
@@ -69,7 +70,7 @@ public void updateServersStatus() {
  }
  }
 
- public boolean ping(HostAddress addr) {
+ public boolean ping(HostAddress addr) throws ClientServerIncompatibleException {
  try {
  Connection connection = new SyncConnection();
  if (enabledSsl) {
@@ -84,7 +85,7 @@ public boolean ping(HostAddress addr) {
  }
  }
 
- public boolean isServersOK() {
+ public boolean isServersOK() throws ClientServerIncompatibleException {
  this.updateServersStatus();
  for (HostAddress addr : addresses) {
  if (serversStatus.get(addr) == S_BAD) {
@@ -94,7 +95,11 @@ public boolean isServersOK() {
  return true;
  }
 
- private void scheduleTask() {
- updateServersStatus();
+ private void scheduleTask() {
+ try {
+ updateServersStatus();
+ } catch (ClientServerIncompatibleException e) {
+ e.printStackTrace();
+ }
  }
 }

client/src/main/java/com/vesoft/nebula/client/graph/net/Session.java

@@ -8,6 +8,7 @@
 
 import com.vesoft.nebula.client.graph.data.HostAddress;
 import com.vesoft.nebula.client.graph.data.ResultSet;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 import com.vesoft.nebula.graph.ExecutionResponse;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -60,7 +61,8 @@ public Session(SyncConnection connection,
  * such as insert ngql `INSERT VERTEX person(name) VALUES "Tom":("Tom");`
  * @return The ResultSet
  */
- public synchronized ResultSet execute(String stmt) throws IOErrorException {
+ public synchronized ResultSet execute(String stmt) throws
+ IOErrorException, ClientServerIncompatibleException {
  if (connection == null) {
  throw new IOErrorException(IOErrorException.E_CONNECT_BROKEN,
  "The session was released, couldn't use again.");
@@ -158,7 +160,8 @@ public synchronized ResultSet execute(String stmt) throws IOErrorException {
  * such as insert ngql `INSERT VERTEX person(name) VALUES "Tom":("Tom");`
  * @return The JSON string
  */
- public synchronized String executeJson(String stmt) throws IOErrorException {
+ public synchronized String executeJson(String stmt) throws
+ IOErrorException, ClientServerIncompatibleException {
  if (connection == null) {
  throw new IOErrorException(IOErrorException.E_CONNECT_BROKEN,
  "The session was released, couldn't use again.");

client/src/main/java/com/vesoft/nebula/client/graph/net/SessionWrapper.java

@@ -7,6 +7,7 @@
 package com.vesoft.nebula.client.graph.net;
 
 import com.vesoft.nebula.client.graph.data.ResultSet;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 import com.vesoft.nebula.client.graph.exception.InvalidSessionException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -25,7 +26,8 @@ public SessionWrapper(Session session) {
  * @param stmt The query sentence.
  * @return The ResultSet.
  */
- public ResultSet execute(String stmt) throws IOErrorException {
+ public ResultSet execute(String stmt)
+ throws IOErrorException, ClientServerIncompatibleException {
  if (!available()) {
  throw new InvalidSessionException();
  }

client/src/main/java/com/vesoft/nebula/client/graph/net/SessionsManager.java

@@ -9,6 +9,7 @@
 import com.vesoft.nebula.client.graph.SessionsManagerConfig;
 import com.vesoft.nebula.client.graph.data.ResultSet;
 import com.vesoft.nebula.client.graph.exception.AuthFailedException;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 import com.vesoft.nebula.client.graph.exception.NotValidConnectionException;
 import java.net.UnknownHostException;
@@ -44,7 +45,8 @@ private void checkConfig() {
  * @return SessionWrapper
  * @throws RuntimeException the exception when get SessionWrapper
  */
- public synchronized SessionWrapper getSessionWrapper() throws RuntimeException {
+ public synchronized SessionWrapper getSessionWrapper() throws RuntimeException,
+ ClientServerIncompatibleException {
  checkClose();
  if (pool == null) {
  init();
@@ -113,7 +115,7 @@ public synchronized void close() {
  isClose = true;
  }
 
- private void init() throws RuntimeException {
+ private void init() throws RuntimeException, ClientServerIncompatibleException {
  try {
  pool = new NebulaPool();
  if (!pool.init(config.getAddresses(), config.getPoolConfig())) {

client/src/main/java/com/vesoft/nebula/client/graph/net/SyncConnection.java

@@ -14,16 +14,20 @@
 import com.facebook.thrift.transport.TTransport;
 import com.facebook.thrift.transport.TTransportException;
 import com.facebook.thrift.utils.StandardCharsets;
+import com.google.common.base.Charsets;
 import com.vesoft.nebula.ErrorCode;
 import com.vesoft.nebula.client.graph.data.CASignedSSLParam;
 import com.vesoft.nebula.client.graph.data.HostAddress;
 import com.vesoft.nebula.client.graph.data.SSLParam;
 import com.vesoft.nebula.client.graph.data.SelfSignedSSLParam;
 import com.vesoft.nebula.client.graph.exception.AuthFailedException;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.graph.exception.IOErrorException;
 import com.vesoft.nebula.graph.AuthResponse;
 import com.vesoft.nebula.graph.ExecutionResponse;
 import com.vesoft.nebula.graph.GraphService;
+import com.vesoft.nebula.graph.VerifyClientVersionReq;
+import com.vesoft.nebula.graph.VerifyClientVersionResp;
 import com.vesoft.nebula.util.SslUtil;
 import java.io.IOException;
 import javax.net.ssl.SSLSocketFactory;
@@ -37,7 +41,8 @@ public class SyncConnection extends Connection {
  private boolean enabledSsl = false;
 
  @Override
- public void open(HostAddress address, int timeout, SSLParam sslParam) throws IOErrorException {
+ public void open(HostAddress address, int timeout, SSLParam sslParam)
+ throws IOErrorException, ClientServerIncompatibleException {
  try {
  SSLSocketFactory sslSocketFactory;
 
@@ -61,24 +66,40 @@ public void open(HostAddress address, int timeout, SSLParam sslParam) throws IOE
  address.getPort()), this.timeout, this.timeout);
  this.protocol = new TCompactProtocol(transport);
  client = new GraphService.Client(protocol);
- } catch (TException e) {
+
+ // check if client version matches server version
+ VerifyClientVersionResp resp =
+ client.verifyClientVersion(new VerifyClientVersionReq());
+ if (resp.error_code != ErrorCode.SUCCEEDED) {
+ client.getInputProtocol().getTransport().close();
+ throw new ClientServerIncompatibleException(new String(resp.getError_msg(),
+ Charsets.UTF_8));
+ }
+ } catch (TException | IOException e) {
  throw new IOErrorException(IOErrorException.E_UNKNOWN, e.getMessage());
- } catch (IOException e) {
- e.printStackTrace();
  }
  }
 
  @Override
- public void open(HostAddress address, int timeout) throws IOErrorException {
+ public void open(HostAddress address, int timeout)
+ throws IOErrorException, ClientServerIncompatibleException {
  try {
- this.enabledSsl = false;
  this.serverAddr = address;
  this.timeout = timeout <= 0 ? Integer.MAX_VALUE : timeout;
  this.transport = new TSocket(
  address.getHost(), address.getPort(), this.timeout, this.timeout);
  this.transport.open();
  this.protocol = new TCompactProtocol(transport);
  client = new GraphService.Client(protocol);
+
+ // check if client version matches server version
+ VerifyClientVersionResp resp =
+ client.verifyClientVersion(new VerifyClientVersionReq());
+ if (resp.error_code != ErrorCode.SUCCEEDED) {
+ client.getInputProtocol().getTransport().close();
+ throw new ClientServerIncompatibleException(new String(resp.getError_msg(),
+ Charsets.UTF_8));
+ }
  } catch (TException e) {
  throw new IOErrorException(IOErrorException.E_UNKNOWN, e.getMessage());
  }
@@ -95,7 +116,7 @@ public void open(HostAddress address, int timeout) throws IOErrorException {
  * @throws IOErrorException if io problem happen
  */
  @Override
- public void reopen() throws IOErrorException {
+ public void reopen() throws IOErrorException, ClientServerIncompatibleException {
  close();
  if (enabledSsl) {
  open(serverAddr, timeout, sslParam);
@@ -105,7 +126,7 @@ public void reopen() throws IOErrorException {
  }
 
  public AuthResult authenticate(String user, String password)
- throws AuthFailedException, IOErrorException {
+  throws AuthFailedException, IOErrorException, ClientServerIncompatibleException {
  try {
  AuthResponse resp = client.authenticate(user.getBytes(), password.getBytes());
  if (resp.error_code != ErrorCode.SUCCEEDED) {
@@ -136,7 +157,7 @@ public AuthResult authenticate(String user, String password)
  }
 
  public ExecutionResponse execute(long sessionID, String stmt)
- throws IOErrorException {
+ throws IOErrorException, ClientServerIncompatibleException {
  try {
  return client.execute(sessionID, stmt.getBytes());
  } catch (TException e) {
@@ -157,7 +178,7 @@ public ExecutionResponse execute(long sessionID, String stmt)
  }
 
  public String executeJson(long sessionID, String stmt)
- throws IOErrorException {
+ throws IOErrorException, ClientServerIncompatibleException {
  try {
  byte[] result = client.executeJson(sessionID, stmt.getBytes());
  return new String(result, StandardCharsets.UTF_8);
@@ -187,8 +208,7 @@ public boolean ping() {
  try {
  execute(0, "YIELD 1;");
  return true;
- } catch (IOErrorException e) {
- e.printStackTrace();
+ } catch (IOErrorException | ClientServerIncompatibleException e) {
  return false;
  }
  }