Improve parameter and error log

apache · Nov 8, 2023 · 5cf0761 · 5cf0761
1 parent 9cbbfcb
commit 5cf0761
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 75 deletions.
diff --git a/...g/apache/shardingsphere/data/pipeline/cdc/client/handler/LoggerExceptionErrorHandler.java b/...g/apache/shardingsphere/data/pipeline/cdc/client/handler/LoggerExceptionErrorHandler.java
diff --git a/...pache/shardingsphere/data/pipeline/cdc/client/handler/RetryStreamingExceptionHandler.java b/...pache/shardingsphere/data/pipeline/cdc/client/handler/RetryStreamingExceptionHandler.java
@@ -31,44 +31,39 @@
  * Retry streaming exception handler.
  */
 @Slf4j
-public class RetryStreamingExceptionHandler implements ExceptionHandler {
+public final class RetryStreamingExceptionHandler implements ExceptionHandler {
 
     private final CDCClient cdcClient;
 
     private final AtomicInteger maxRetryTimes = new AtomicInteger(0);
 
     private final int retryIntervalMills;
 
-    private final int retryTimes;
+    private final AtomicInteger retryTimes = new AtomicInteger(0);
 
     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());
+        log.error("Catch exception: ", throwable);
         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;
-        }
+        retryTimes.incrementAndGet();
         ClientConnectionContext connectionContext = ctx.channel().attr(ClientConnectionContext.CONTEXT_KEY).get();
-        if (retryTimes > maxRetryTimes.get()) {
-            log.warn("Retry times exceed 5, stop streaming");
+        if (retryTimes.get() > maxRetryTimes.get()) {
+            log.warn("Stop try to reconnect, stop streaming ids: {}", connectionContext.getStreamingIds());
             connectionContext.getStreamingIds().forEach(each -> CompletableFuture.runAsync(() -> cdcClient.stopStreaming(each)));
             return;
         }
         TimeUnit.MILLISECONDS.sleep(retryIntervalMills);
-        log.info("Retry to restart streaming, retry count: {}", maxRetryTimes.get());
+        log.info("Retry to restart streaming, retry times: {}", retryTimes.get());
         connectionContext.getStreamingIds().forEach(each -> CompletableFuture.runAsync(() -> cdcClient.restartStreaming(each)));
     }
 }
diff --git a/...t/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/client/example/Bootstrap.java b/...t/src/test/java/org/apache/shardingsphere/data/pipeline/cdc/client/example/Bootstrap.java
@@ -21,7 +21,6 @@
 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.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;
@@ -42,24 +41,13 @@ 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";
-        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++;
+        try (CDCClient cdcClient = new CDCClient(new CDCClientConfiguration(address, 33071, 10000))) {
+            cdcClient.connect(records -> log.info("records: {}", records), new RetryStreamingExceptionHandler(cdcClient, 5, 5000),
+                    (ctx, result) -> log.error("Server error: {}", result.getErrorMessage()));
+            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();
         }
     }
 }
diff --git a/...ne/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java b/...ne/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/cdc/CDCE2EIT.java
@@ -23,7 +23,7 @@
 import org.apache.shardingsphere.data.pipeline.cdc.api.job.type.CDCJobType;
 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.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;
@@ -171,8 +171,7 @@ private void startCDCClient(final PipelineContainerComposer containerComposer, f
         DataSourceRecordConsumer recordConsumer = new DataSourceRecordConsumer(dataSource, containerComposer.getDatabaseType());
         String schema = dialectDatabaseMetaData.isSchemaAvailable() ? "test" : "";
         CDCClient cdcClient = new CDCClient(new CDCClientConfiguration("localhost", containerComposer.getContainerComposer().getProxyCDCPort(), 5000));
-        LoggerExceptionErrorHandler handler = new LoggerExceptionErrorHandler();
-        cdcClient.connect(recordConsumer, handler, handler);
+        cdcClient.connect(recordConsumer, new RetryStreamingExceptionHandler(cdcClient, 5, 5000), (ctx, result) -> log.error("Server error: {}", result.getErrorMessage()));
         cdcClient.login(new CDCLoginParameter(ProxyContainerConstants.USERNAME, ProxyContainerConstants.PASSWORD));
         // TODO add full=false test case later
         cdcClient.startStreaming(new StartStreamingParameter("sharding_db", new HashSet<>(Arrays.asList(SchemaTable.newBuilder().setTable(SOURCE_TABLE_NAME).setSchema(schema).build(),