Improve CDCImporter process FinishedRecord, and advise CDC client

kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/CDCClient.java

@@ -35,6 +35,8 @@
 import org.apache.shardingsphere.data.pipeline.cdc.client.constant.ClientConnectionStatus;
 import org.apache.shardingsphere.data.pipeline.cdc.client.context.ClientConnectionContext;
 import org.apache.shardingsphere.data.pipeline.cdc.client.handler.CDCRequestHandler;
+import org.apache.shardingsphere.data.pipeline.cdc.client.handler.ExceptionHandler;
+import org.apache.shardingsphere.data.pipeline.cdc.client.handler.ServerErrorResultHandler;
 import org.apache.shardingsphere.data.pipeline.cdc.client.parameter.CDCLoginParameter;
 import org.apache.shardingsphere.data.pipeline.cdc.client.parameter.StartStreamingParameter;
 import org.apache.shardingsphere.data.pipeline.cdc.client.util.RequestIdUtils;
@@ -49,6 +51,10 @@
 import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StopStreamingRequestBody;
 import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StreamDataRequestBody;
 import org.apache.shardingsphere.data.pipeline.cdc.protocol.response.CDCResponse;
+import org.apache.shardingsphere.data.pipeline.cdc.protocol.response.DataRecordResult.Record;
+
+import java.util.List;
+import java.util.function.Consumer;
 
 /**
  * CDC client.
@@ -78,9 +84,13 @@ private void validateParameter(final CDCClientConfiguration parameter) {
 
     /**
      * Connect.
+     *
+     * @param dataConsumer data consumer
+     * @param exceptionHandler exception handler
+     * @param errorResultHandler error result handler
      */
     @SneakyThrows(InterruptedException.class)
-    public void connect() {
+    public void connect(final Consumer<List<Record>> dataConsumer, final ExceptionHandler exceptionHandler, final ServerErrorResultHandler errorResultHandler) {
         Bootstrap bootstrap = new Bootstrap();
         group = new NioEventLoopGroup(1);
         bootstrap.channel(NioSocketChannel.class)
@@ -95,7 +105,7 @@ protected void initChannel(final NioSocketChannel channel) {
                         channel.pipeline().addLast(new ProtobufDecoder(CDCResponse.getDefaultInstance()));
                         channel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                         channel.pipeline().addLast(new ProtobufEncoder());
-                        channel.pipeline().addLast(new CDCRequestHandler(config.getDataConsumer(), config.getExceptionHandler()));
+                        channel.pipeline().addLast(new CDCRequestHandler(dataConsumer, exceptionHandler, errorResultHandler));
                     }
                 });
         channel = bootstrap.connect(config.getAddress(), config.getPort()).sync().channel();
@@ -165,9 +175,6 @@ public String startStreaming(final StartStreamingParameter parameter) {
      * @param streamingId streaming id
      */
     public void restartStreaming(final String streamingId) {
-        if (checkStreamingIdExist(streamingId)) {
-            stopStreaming(streamingId);
-        }
         String requestId = RequestIdUtils.generateRequestId();
         StartStreamingRequestBody body = StartStreamingRequestBody.newBuilder().setStreamingId(streamingId).build();
         CDCRequest request = CDCRequest.newBuilder().setRequestId(requestId).setType(Type.START_STREAMING).setStartStreamingRequestBody(body).build();
@@ -179,16 +186,6 @@ public void restartStreaming(final String streamingId) {
         log.info("Restart streaming success, streaming id: {}", streamingId);
     }
 
-    private boolean checkStreamingIdExist(final String streamingId) {
-        checkChannelActive();
-        ClientConnectionContext connectionContext = channel.attr(ClientConnectionContext.CONTEXT_KEY).get();
-        if (null == connectionContext) {
-            log.warn("The connection context not exist");
-            return true;
-        }
-        return connectionContext.getStreamingIds().contains(streamingId);
-    }
-
     /**
      * Stop streaming.
      *

kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/config/CDCClientConfiguration.java

@@ -19,11 +19,6 @@
 
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.data.pipeline.cdc.client.handler.ExceptionHandler;
-import org.apache.shardingsphere.data.pipeline.cdc.protocol.response.DataRecordResult.Record;
-
-import java.util.List;
-import java.util.function.Consumer;
 
 /**
  * CDC client configuration.
@@ -36,17 +31,5 @@ public final class CDCClientConfiguration {
 
     private final int port;
 
-    private final Consumer<List<Record>> dataConsumer;
-
-    private final ExceptionHandler exceptionHandler;
-
     private final int timeoutMills;
-
-    public CDCClientConfiguration(final String address, final int port, final Consumer<List<Record>> dataConsumer, final ExceptionHandler exceptionHandler) {
-        this.address = address;
-        this.port = port;
-        this.dataConsumer = dataConsumer;
-        this.exceptionHandler = exceptionHandler;
-        this.timeoutMills = 5000;
-    }
 }

kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/handler/CDCRequestHandler.java

@@ -50,6 +50,8 @@ public final class CDCRequestHandler extends ChannelInboundHandlerAdapter {
 
     private final ExceptionHandler exceptionHandler;
 
+    private final ServerErrorResultHandler errorResultHandler;
+
     @Override
     public void channelRegistered(final ChannelHandlerContext ctx) {
         ClientConnectionContext context = new ClientConnectionContext();
@@ -78,7 +80,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
                 future.countDown();
                 requestType = future.getRequestType();
             }
-            exceptionHandler.handleServerException(new ServerErrorResult(response.getErrorCode(), response.getErrorMessage(), requestType));
+            errorResultHandler.handleServerError(ctx, new ServerErrorResult(response.getErrorCode(), response.getErrorMessage(), requestType));
             responseFuture.ifPresent(future -> {
                 future.setErrorCode(response.getErrorCode());
                 future.setErrorMessage(response.getErrorMessage());
@@ -113,6 +115,6 @@ private void processDataRecords(final ChannelHandlerContext ctx, final DataRecor
 
     @Override
     public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
-        exceptionHandler.handleSocketException(cause);
+        exceptionHandler.handleException(ctx, cause);
     }
 }

kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/handler/ExceptionHandler.java

@@ -17,24 +17,18 @@
 
 package org.apache.shardingsphere.data.pipeline.cdc.client.handler;
 
-import org.apache.shardingsphere.data.pipeline.cdc.client.util.ServerErrorResult;
+import io.netty.channel.ChannelHandlerContext;
 
 /**
  * Exception handler.
  */
 public interface ExceptionHandler {
 
     /**
-     * Handle server exception.
-     *
-     * @param result error result
-     */
-    void handleServerException(ServerErrorResult result);
-
-    /**
-     * Handle socket exception.
+     * Handle exception.
      *
+     * @param ctx channel handler context
      * @param throwable throwable
      */
-    void handleSocketException(Throwable throwable);
+    void handleException(ChannelHandlerContext ctx, Throwable throwable);
 }

kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/handler/LoggerExceptionHandler.java → kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/handler/LoggerExceptionErrorHandler.java

@@ -17,24 +17,25 @@
 
 package org.apache.shardingsphere.data.pipeline.cdc.client.handler;
 
+import io.netty.channel.ChannelHandlerContext;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.shardingsphere.data.pipeline.cdc.client.util.ServerErrorResult;
 
 /**
- * Logger exception handler, just print error.
+ * Logger exception error handler.
  */
 @RequiredArgsConstructor
 @Slf4j
-public class LoggerExceptionHandler implements ExceptionHandler {
+public final class LoggerExceptionErrorHandler implements ExceptionHandler, ServerErrorResultHandler {
 
     @Override
-    public void handleServerException(final ServerErrorResult result) {
+    public void handleServerError(final ChannelHandlerContext ctx, final ServerErrorResult result) {
         log.error("Server error, code: {}, message: {}", result.getErrorCode(), result.getErrorMessage());
     }
 
     @Override
-    public void handleSocketException(final Throwable throwable) {
-        log.error("Socket error: ", throwable);
+    public void handleException(final ChannelHandlerContext ctx, final Throwable throwable) {
+        log.error("Exception error: ", throwable);
     }
 }

kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/handler/RetryStreamingExceptionHandler.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.cdc.client.handler;
+
+import io.netty.channel.ChannelHandlerContext;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.data.pipeline.cdc.client.CDCClient;
+import org.apache.shardingsphere.data.pipeline.cdc.client.context.ClientConnectionContext;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Retry streaming exception handler.
+ */
+@Slf4j
+public class RetryStreamingExceptionHandler implements ExceptionHandler {
+
+    private final CDCClient cdcClient;
+
+    private final AtomicInteger maxRetryTimes = new AtomicInteger(0);
+
+    private final int retryIntervalMills;
+
+    private final int retryTimes;
+
+    public RetryStreamingExceptionHandler(final CDCClient cdcClient, final int maxRetryTimes, final int retryIntervalMills) {
+        this.cdcClient = cdcClient;
+        this.maxRetryTimes.set(maxRetryTimes);
+        this.retryIntervalMills = retryIntervalMills;
+        retryTimes = 0;
+    }
+
+    @Override
+    public void handleException(final ChannelHandlerContext ctx, final Throwable throwable) {
+        log.error("Socket error: {}", throwable.getMessage());
+        reconnect(ctx);
+    }
+
+    @SneakyThrows(InterruptedException.class)
+    private void reconnect(final ChannelHandlerContext ctx) {
+        maxRetryTimes.incrementAndGet();
+        if (null == cdcClient) {
+            log.warn("CDC client is null, could not retry");
+            return;
+        }
+        ClientConnectionContext connectionContext = ctx.channel().attr(ClientConnectionContext.CONTEXT_KEY).get();
+        if (retryTimes > maxRetryTimes.get()) {
+            log.warn("Retry times exceed 5, stop streaming");
+            connectionContext.getStreamingIds().forEach(each -> CompletableFuture.runAsync(() -> cdcClient.stopStreaming(each)));
+            return;
+        }
+        TimeUnit.MILLISECONDS.sleep(retryIntervalMills);
+        log.info("Retry to restart streaming, retry count: {}", maxRetryTimes.get());
+        connectionContext.getStreamingIds().forEach(each -> CompletableFuture.runAsync(() -> cdcClient.restartStreaming(each)));
+    }
+}

kernel/data-pipeline/scenario/cdc/client/src/main/java/org/apache/shardingsphere/data/pipeline/cdc/client/handler/ServerErrorResultHandler.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.data.pipeline.cdc.client.handler;
+
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.shardingsphere.data.pipeline.cdc.client.util.ServerErrorResult;
+
+/**
+ * Server error result handler.
+ */
+public interface ServerErrorResultHandler {
+
+    /**
+     * Handle server ERROR.
+     *
+     * @param ctx channel handler context
+     * @param result error result
+     */
+    void handleServerError(ChannelHandlerContext ctx, ServerErrorResult result);
+}

kernel/data-pipeline/scenario/cdc/client/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/client/example/Bootstrap.java

@@ -21,7 +21,8 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.shardingsphere.data.pipeline.cdc.client.CDCClient;
 import org.apache.shardingsphere.data.pipeline.cdc.client.config.CDCClientConfiguration;
-import org.apache.shardingsphere.data.pipeline.cdc.client.handler.LoggerExceptionHandler;
+import org.apache.shardingsphere.data.pipeline.cdc.client.handler.LoggerExceptionErrorHandler;
+import org.apache.shardingsphere.data.pipeline.cdc.client.handler.RetryStreamingExceptionHandler;
 import org.apache.shardingsphere.data.pipeline.cdc.client.parameter.CDCLoginParameter;
 import org.apache.shardingsphere.data.pipeline.cdc.client.parameter.StartStreamingParameter;
 import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StreamDataRequestBody.SchemaTable;
@@ -41,13 +42,24 @@ public static void main(final String[] args) {
         // Pay attention to the time zone, to avoid the problem of incorrect time zone, it is best to ensure that the time zone of the program is consistent with the time zone of the database server
         // TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
         String address = "127.0.0.1";
-        CDCClientConfiguration clientConfig = new CDCClientConfiguration(address, 33071, records -> log.info("records: {}", records), new LoggerExceptionHandler());
-        try (CDCClient cdcClient = new CDCClient(clientConfig)) {
-            cdcClient.connect();
-            cdcClient.login(new CDCLoginParameter("root", "root"));
-            String streamingId = cdcClient.startStreaming(new StartStreamingParameter("sharding_db", Collections.singleton(SchemaTable.newBuilder().setTable("t_order").build()), true));
-            log.info("Streaming id={}", streamingId);
-            cdcClient.await();
+        int reconnectCount = 0;
+        int maxReconnectCount = 5;
+        while (reconnectCount < maxReconnectCount) {
+            try (CDCClient cdcClient = new CDCClient(new CDCClientConfiguration(address, 33071, 5000))) {
+                LoggerExceptionErrorHandler loggerExceptionErrorHandler = new LoggerExceptionErrorHandler();
+                cdcClient.connect(records -> log.info("records: {}", records), new RetryStreamingExceptionHandler(cdcClient, 5, 5000), loggerExceptionErrorHandler);
+                cdcClient.login(new CDCLoginParameter("root", "root"));
+                String streamingId = cdcClient.startStreaming(new StartStreamingParameter("sharding_db", Collections.singleton(SchemaTable.newBuilder().setTable("t_order123").build()), true));
+                log.info("Streaming id={}", streamingId);
+                cdcClient.await();
+                // CHECKSTYLE:OFF
+            } catch (final Exception ex) {
+                // CHECKSTYLE:ON
+                log.error("Exception occur: {}", ex.getMessage());
+            }
+            Thread.sleep(5000);
+            log.info("Reconnect count: {}", reconnectCount);
+            reconnectCount++;
         }
     }
 }