[IOTDB-4135] Merge thrift-sync into ClientRPC (apache#7004)

Author: Cpaulyz

antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -65,8 +65,7 @@ utilityStatement
     | loadConfiguration | loadTimeseries | loadFile | removeFile | unloadFile;
 
 syncStatement
-    : startPipeServer | stopPipeServer | showPipeServer
-    | createPipeSink | showPipeSinkType | showPipeSink | dropPipeSink
+    : createPipeSink | showPipeSinkType | showPipeSink | dropPipeSink
     | createPipe | showPipe | stopPipe | startPipe | dropPipe;
 
 /**
@@ -764,18 +763,6 @@ syncAttributeClauses
     : attributePair (COMMA attributePair)*
     ;
 
-// sync receiver
-startPipeServer
-    : START PIPESERVER
-    ;
-
-stopPipeServer
-    : STOP PIPESERVER
-    ;
-
-showPipeServer
-    : SHOW PIPESERVER
-    ;
 
 /**
  * 7. Common Clauses

antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4

@@ -379,10 +379,6 @@ PIPES
     : P I P E S
     ;
 
-PIPESERVER
-    : P I P E S E R V E R
-    ;
-
 PIPESINK
     : P I P E S I N K
     ;

integration/src/test/java/org/apache/iotdb/db/integration/sync/IoTDBSyncReceiverIT.java

@@ -21,19 +21,17 @@
 import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.engine.modification.Deletion;
-import org.apache.iotdb.db.exception.sync.PipeServerException;
 import org.apache.iotdb.db.qp.physical.PhysicalPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateAlignedTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
 import org.apache.iotdb.db.qp.physical.sys.SetStorageGroupPlan;
-import org.apache.iotdb.db.sync.SyncService;
 import org.apache.iotdb.db.sync.pipedata.DeletionPipeData;
 import org.apache.iotdb.db.sync.pipedata.PipeData;
 import org.apache.iotdb.db.sync.pipedata.SchemaPipeData;
 import org.apache.iotdb.db.sync.pipedata.TsFilePipeData;
 import org.apache.iotdb.db.sync.sender.pipe.Pipe;
 import org.apache.iotdb.db.sync.sender.pipe.TsFilePipe;
-import org.apache.iotdb.db.sync.transport.client.IoTDBSInkTransportClient;
+import org.apache.iotdb.db.sync.transport.client.IoTDBSinkTransportClient;
 import org.apache.iotdb.db.utils.EnvironmentUtils;
 import org.apache.iotdb.itbase.category.LocalStandaloneTest;
 import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
@@ -51,7 +49,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
-import java.net.Socket;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -71,7 +68,7 @@ public class IoTDBSyncReceiverIT {
   String remoteIp1;
   long createdTime1 = System.currentTimeMillis();
   String showPipeSql = "SHOW PIPE";
-  IoTDBSInkTransportClient client;
+  IoTDBSinkTransportClient client;
 
   @Before
   public void setUp() throws Exception {
@@ -94,15 +91,9 @@ public void setUp() throws Exception {
     FileUtils.moveDirectory(srcDir, tmpDir);
     EnvironmentUtils.cleanEnv();
     EnvironmentUtils.envSetUp();
-    try {
-      SyncService.getInstance().startPipeServer(true);
-      new Socket("localhost", 6670).close();
-    } catch (Exception e) {
-      Assert.fail("Failed to start pipe server because " + e.getMessage());
-    }
     Pipe pipe = new TsFilePipe(createdTime1, pipeName1, null, 0, false);
     remoteIp1 = "127.0.0.1";
-    client = new IoTDBSInkTransportClient(pipe, remoteIp1, 6670, "127.0.0.1");
+    client = new IoTDBSinkTransportClient(pipe, remoteIp1, 6667, "127.0.0.1");
     client.handshake();
   }
 
@@ -120,16 +111,6 @@ public void tearDown() throws Exception {
     EnvironmentUtils.cleanEnv();
   }
 
-  @Test
-  public void testStopPipeServer() {
-    logger.info("testStopPipeServerCheck");
-    try {
-      SyncService.getInstance().stopPipeServer();
-    } catch (PipeServerException e) {
-      Assert.fail("Can not stop pipe server");
-    }
-  }
-
   @Test
   public void testReceiveDataAndLoad() {
     logger.info("testReceiveDataAndLoad");

node-commons/pom.xml

@@ -77,11 +77,6 @@
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.apache.iotdb</groupId>
-            <artifactId>iotdb-thrift</artifactId>
-            <version>${project.version}</version>
-        </dependency>
         <dependency>
             <groupId>org.apache.iotdb</groupId>
             <artifactId>iotdb-thrift-confignode</artifactId>

node-commons/src/main/java/org/apache/iotdb/commons/sync/SyncConstant.java

@@ -72,12 +72,6 @@ public class SyncConstant {
   public static final int DATA_CHUNK_SIZE =
       Math.min(16 * 1024 * 1024, RpcUtils.THRIFT_FRAME_MAX_SIZE);
 
-  public static final int SUCCESS_CODE = 1;
-  public static final int ERROR_CODE = -1;
-  public static final int REBASE_CODE = -2;
-  public static final int RETRY_CODE = -3;
-  public static final int CONFLICT_CODE = -4;
-
   /** receiver */
   public static final String RECEIVER_DIR_NAME = "receiver";
 

node-commons/src/main/java/org/apache/iotdb/commons/sync/SyncPathUtil.java

@@ -20,7 +20,7 @@
 package org.apache.iotdb.commons.sync;
 
 import org.apache.iotdb.commons.conf.CommonDescriptor;
-import org.apache.iotdb.service.transport.thrift.IdentityInfo;
+import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo;
 
 import java.io.File;
 import java.io.IOException;
@@ -105,12 +105,12 @@ public static String getReceiverFileDataDir(String pipeName, String remoteIp, lo
         + SyncConstant.FILE_DATA_DIR_NAME;
   }
 
-  public static String getFileDataDirPath(IdentityInfo identityInfo) {
+  public static String getFileDataDirPath(TSyncIdentityInfo identityInfo) {
     return SyncPathUtil.getReceiverFileDataDir(
         identityInfo.getPipeName(), identityInfo.getAddress(), identityInfo.getCreateTime());
   }
 
-  public static String getPipeLogDirPath(IdentityInfo identityInfo) {
+  public static String getPipeLogDirPath(TSyncIdentityInfo identityInfo) {
     return SyncPathUtil.getReceiverPipeLogDir(
         identityInfo.getPipeName(), identityInfo.getAddress(), identityInfo.getCreateTime());
   }

pom.xml

@@ -85,7 +85,6 @@
         <module>thrift-commons</module>
         <module>thrift-confignode</module>
         <module>thrift-multi-leader-consensus</module>
-        <module>thrift-sync</module>
         <module>thrift-influxdb</module>
         <module>service-rpc</module>
         <module>jdbc</module>

server/src/assembly/resources/conf/iotdb-datanode.properties

@@ -675,10 +675,6 @@ timestamp_precision=ms
 ####################
 ### PIPE Server Configuration
 ####################
-# PIPE server port to listen
-# Datatype: int
-# pipe_server_port=6670
-
 # White IP list of Sync client.
 # Please use the form of network segment to present the range of IP, for example: 192.168.0.0/16
 # If there are more than one IP segment, please separate them by commas

server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -501,9 +501,6 @@ public class IoTDBConfig {
    */
   private int externalSortThreshold = 1000;
 
-  /** If this IoTDB instance is a receiver of sync, set the server port. */
-  private int pipeServerPort = 6670;
-
   /** White list for sync */
   private String ipWhiteList = "0.0.0.0/0";
 
@@ -1441,14 +1438,6 @@ public void setmRemoteSchemaCacheSize(int mRemoteSchemaCacheSize) {
     this.mRemoteSchemaCacheSize = mRemoteSchemaCacheSize;
   }
 
-  public int getPipeServerPort() {
-    return pipeServerPort;
-  }
-
-  public void setPipeServerPort(int pipeServerPort) {
-    this.pipeServerPort = pipeServerPort;
-  }
-
   public int getMaxNumberOfSyncFileRetry() {
     return maxNumberOfSyncFileRetry;
   }

server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -417,11 +417,6 @@ private void loadProps() {
               properties.getProperty(
                   "session_timeout_threshold",
                   Integer.toString(conf.getSessionTimeoutThreshold()))));
-      conf.setPipeServerPort(
-          Integer.parseInt(
-              properties
-                  .getProperty("pipe_server_port", Integer.toString(conf.getPipeServerPort()))
-                  .trim()));
       conf.setMaxNumberOfSyncFileRetry(
           Integer.parseInt(
               properties
@@ -1417,10 +1412,6 @@ public void loadHotModifiedProps(Properties properties) throws QueryProcessExcep
                   String.valueOf(conf.getSelectIntoInsertTabletPlanRowLimit()))));
 
       // update sync config
-      conf.setPipeServerPort(
-          Integer.parseInt(
-              properties.getProperty(
-                  "pipe_server_port", String.valueOf(conf.getPipeServerPort()))));
       conf.setMaxNumberOfSyncFileRetry(
           Integer.parseInt(
               properties