add version verification

client/src/main/java/com/vesoft/nebula/client/meta/MetaClient.java

@@ -10,9 +10,11 @@
 import com.facebook.thrift.protocol.TCompactProtocol;
 import com.facebook.thrift.transport.TSocket;
 import com.facebook.thrift.transport.TTransportException;
+import com.google.common.base.Charsets;
 import com.vesoft.nebula.ErrorCode;
 import com.vesoft.nebula.HostAddr;
 import com.vesoft.nebula.client.graph.data.HostAddress;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.meta.exception.ExecuteFailedException;
 import com.vesoft.nebula.meta.EdgeItem;
 import com.vesoft.nebula.meta.GetEdgeReq;
@@ -39,6 +41,8 @@
 import com.vesoft.nebula.meta.Schema;
 import com.vesoft.nebula.meta.SpaceItem;
 import com.vesoft.nebula.meta.TagItem;
+import com.vesoft.nebula.meta.VerifyClientVersionReq;
+import com.vesoft.nebula.meta.VerifyClientVersionResp;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -85,28 +89,41 @@ public MetaClient(List<HostAddress> addresses, int timeout, int connectionRetry,
  this.addresses = addresses;
  }
 
- public void connect() throws TException {
+ public void connect()
+ throws TException, ClientServerIncompatibleException {
  doConnect();
  }
 
  /**
  * connect nebula meta server
  */
- private void doConnect() throws TTransportException {
+ private void doConnect()
+ throws TTransportException, ClientServerIncompatibleException {
  Random random = new Random(System.currentTimeMillis());
  int position = random.nextInt(addresses.size());
  HostAddress address = addresses.get(position);
  getClient(address.getHost(), address.getPort());
  }
 
- private void getClient(String host, int port) throws TTransportException {
+ private void getClient(String host, int port)
+ throws TTransportException, ClientServerIncompatibleException {
  transport = new TSocket(host, port, timeout, timeout);
  transport.open();
  protocol = new TCompactProtocol(transport);
  client = new MetaService.Client(protocol);
+
+ // check if client version matches server version
+ VerifyClientVersionResp resp =
+ client.verifyClientVersion(new VerifyClientVersionReq());
+ if (resp.getCode() != ErrorCode.SUCCEEDED) {
+ client.getInputProtocol().getTransport().close();
+ throw new ClientServerIncompatibleException(new String(resp.getError_msg(),
+ Charsets.UTF_8));
+ }
  }
 
- private void freshClient(HostAddr leader) throws TTransportException {
+ private void freshClient(HostAddr leader)
+ throws TTransportException, ClientServerIncompatibleException {
  close();
  getClient(leader.getHost(), leader.getPort());
  }
@@ -125,7 +142,8 @@ public void close() {
  *
  * @return
  */
- public synchronized List<IdName> getSpaces() throws TException, ExecuteFailedException {
+ public synchronized List<IdName> getSpaces()
+ throws TException, ExecuteFailedException, ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
  ListSpacesReq request = new ListSpacesReq();
  ListSpacesResp response = null;
@@ -158,7 +176,7 @@ public synchronized List<IdName> getSpaces() throws TException, ExecuteFailedExc
  * @return SpaceItem
  */
  public synchronized SpaceItem getSpace(String spaceName) throws TException,
- ExecuteFailedException {
+ ExecuteFailedException, ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
  GetSpaceReq request = new GetSpaceReq();
  request.setSpace_name(spaceName.getBytes());
@@ -192,7 +210,7 @@ public synchronized SpaceItem getSpace(String spaceName) throws TException,
  * @return TagItem list
  */
  public synchronized List<TagItem> getTags(String spaceName)
- throws TException, ExecuteFailedException {
+ throws TException, ExecuteFailedException, ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
 
  int spaceID = getSpace(spaceName).space_id;
@@ -229,7 +247,7 @@ public synchronized List<TagItem> getTags(String spaceName)
  * @return Schema
  */
  public synchronized Schema getTag(String spaceName, String tagName)
- throws TException, ExecuteFailedException {
+ throws TException, ExecuteFailedException, ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
  GetTagReq request = new GetTagReq();
  int spaceID = getSpace(spaceName).getSpace_id();
@@ -268,7 +286,7 @@ public synchronized Schema getTag(String spaceName, String tagName)
  * @return EdgeItem list
  */
  public synchronized List<EdgeItem> getEdges(String spaceName)
- throws TException, ExecuteFailedException {
+ throws TException, ExecuteFailedException, ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
  int spaceID = getSpace(spaceName).getSpace_id();
  ListEdgesReq request = new ListEdgesReq(spaceID);
@@ -303,7 +321,7 @@ public synchronized List<EdgeItem> getEdges(String spaceName)
  * @return Schema
  */
  public synchronized Schema getEdge(String spaceName, String edgeName)
- throws TException, ExecuteFailedException {
+ throws TException, ExecuteFailedException, ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
  GetEdgeReq request = new GetEdgeReq();
  int spaceID = getSpace(spaceName).getSpace_id();
@@ -343,7 +361,7 @@ public synchronized Schema getEdge(String spaceName, String edgeName)
  * @return
  */
  public synchronized Map<Integer, List<HostAddr>> getPartsAlloc(String spaceName)
- throws ExecuteFailedException, TException {
+ throws ExecuteFailedException, TException, ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
  GetPartsAllocReq request = new GetPartsAllocReq();
  int spaceID = getSpace(spaceName).getSpace_id();
@@ -375,7 +393,7 @@ public synchronized Map<Integer, List<HostAddr>> getPartsAlloc(String spaceName)
  /**
  * get all Storaged servers
  */
- public synchronized Set<HostAddr> listHosts() {
+ public synchronized Set<HostAddr> listHosts() throws ClientServerIncompatibleException {
  int retry = RETRY_TIMES;
  ListHostsReq request = new ListHostsReq();
  request.setType(ListHostType.STORAGE);

client/src/main/java/com/vesoft/nebula/client/meta/MetaManager.java

@@ -10,6 +10,7 @@
 import com.google.common.collect.Maps;
 import com.vesoft.nebula.HostAddr;
 import com.vesoft.nebula.client.graph.data.HostAddress;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.meta.exception.ExecuteFailedException;
 import com.vesoft.nebula.meta.EdgeItem;
 import com.vesoft.nebula.meta.IdName;
@@ -49,7 +50,8 @@ private class SpaceInfo {
  /**
  * init the meta info cache
  */
- public MetaManager(List<HostAddress> address) throws TException {
+ public MetaManager(List<HostAddress> address)
+ throws TException, ClientServerIncompatibleException {
  metaClient = new MetaClient(address);
  metaClient.connect();
  fillMetaInfo();
@@ -120,7 +122,7 @@ private void fillMetaInfo() {
  } finally {
  lock.writeLock().unlock();
  }
- } catch (TException | ExecuteFailedException e) {
+ } catch (TException | ExecuteFailedException | ClientServerIncompatibleException e) {
  LOGGER.error(e.getMessage());
  }
  }
@@ -330,7 +332,13 @@ public void updateLeader(String spaceName, int part, HostAddr newLeader)
  * get all storage addresses
  */
  public Set<HostAddr> listHosts() {
- Set<HostAddr> hosts = metaClient.listHosts();
+ Set<HostAddr> hosts;
+ try {
+ hosts = metaClient.listHosts();
+ } catch (ClientServerIncompatibleException e) {
+ LOGGER.error("client version does not match server version");
+ return new HashSet<>();
+ }
  if (hosts == null) {
  return new HashSet<>();
  }

client/src/test/java/com/vesoft/nebula/client/meta/TestMetaClient.java

@@ -7,6 +7,7 @@
 package com.vesoft.nebula.client.meta;
 
 import com.facebook.thrift.TException;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.client.meta.exception.ExecuteFailedException;
 import com.vesoft.nebula.client.util.ProcessUtil;
 import com.vesoft.nebula.meta.EdgeItem;
@@ -40,7 +41,7 @@ private void connect() {
  metaClient = new MetaClient(address, port);
  try {
  metaClient.connect();
- } catch (TException e) {
+ } catch (TException | ClientServerIncompatibleException e) {
  e.printStackTrace();
  assert (false);
  }
@@ -51,7 +52,7 @@ public void testFailConnect() {
  MetaClient client = new MetaClient(address, port);
  try {
  client.connect();
- } catch (TException e) {
+ } catch (TException | ClientServerIncompatibleException e) {
  assert (true);
  }
  }
@@ -61,7 +62,7 @@ public void testGetSpaces() {
  List<IdName> spaces = metaClient.getSpaces();
  assert (spaces.size() >= 1);
  assert (metaClient.getSpace("testMeta") != null);
- } catch (TException | ExecuteFailedException e) {
+ } catch (TException | ExecuteFailedException | ClientServerIncompatibleException e) {
  LOGGER.error(e.getMessage());
  e.printStackTrace();
  assert (false);
@@ -73,7 +74,7 @@ public void testGetTags() {
  List<TagItem> tags = metaClient.getTags("testMeta");
  Assert.assertTrue(tags.size() >= 1);
  assert (metaClient.getTag("testMeta", "person") != null);
- } catch (TException | ExecuteFailedException e) {
+ } catch (TException | ExecuteFailedException | ClientServerIncompatibleException e) {
  e.printStackTrace();
  assert (false);
  }
@@ -84,7 +85,7 @@ public void testGetEdges() {
  List<EdgeItem> edges = metaClient.getEdges("testMeta");
  Assert.assertTrue(edges.size() >= 1);
  assert (metaClient.getEdge("testMeta", "friend") != null);
- } catch (TException | ExecuteFailedException e) {
+ } catch (TException | ExecuteFailedException | ClientServerIncompatibleException e) {
  e.printStackTrace();
  assert (false);
  }
@@ -93,13 +94,13 @@ public void testGetEdges() {
  public void testGetPartsAlloc() {
  try {
  assert (metaClient.getPartsAlloc("testMeta").size() == 10);
- } catch (ExecuteFailedException | TException e) {
+ } catch (ExecuteFailedException | TException | ClientServerIncompatibleException e) {
  e.printStackTrace();
  assert (false);
  }
  }
 
- public void testListHosts() {
+ public void testListHosts() throws ClientServerIncompatibleException {
  if (metaClient == null) {
  metaClient = new MetaClient(address, port);
  }
@@ -122,7 +123,12 @@ public void testListOnlineHosts() {
  if (metaClient == null) {
  metaClient = new MetaClient(address, port);
  }
- assert (metaClient.listHosts().size() == 2);
+ try {
+ assert (metaClient.listHosts().size() == 2);
+ } catch (ClientServerIncompatibleException e) {
+ LOGGER.error("client version does not match server version", e);
+ assert (false);
+ }
 
  try {
  runtime.exec("docker start nebula-docker-compose_storaged0_1")

client/src/test/java/com/vesoft/nebula/client/meta/TestMetaManager.java

@@ -8,6 +8,7 @@
 
 import com.vesoft.nebula.HostAddr;
 import com.vesoft.nebula.client.graph.data.HostAddress;
+import com.vesoft.nebula.client.graph.exception.ClientServerIncompatibleException;
 import com.vesoft.nebula.meta.EdgeItem;
 import com.vesoft.nebula.meta.SpaceItem;
 import com.vesoft.nebula.meta.TagItem;
@@ -97,7 +98,7 @@ public void testGetSpaceParts() {
  Assert.assertEquals(hostAddr.port, 4400);
  }
 
- public void testMultiVersionSchema() {
+ public void testMultiVersionSchema() throws ClientServerIncompatibleException {
  MockNebulaGraph.createMultiVersionTagAndEdge();
  metaManager.close();
  metaManager = new MetaManager(