feat(scheduler): 恢复调度时间点

pom.xml

@@ -8,7 +8,7 @@
     <artifactId>hudi-service</artifactId>
     <version>1.0.0-SNAPSHOT</version>
     <packaging>pom</packaging>
-    <description>Hudi服务应用集合项目详细代码目录</description>
+    <description>Hudi服务模块系列应用</description>
     <modules>
         <module>service-common</module>
         <module>service-dependencies</module>

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/configuration/ScheduleStrategyProvider.java

@@ -26,9 +26,9 @@ public class ScheduleStrategyProvider {
         return Lists.immutable.of(
                 ScheduleStrategyImpl.simple(false, "distribute_schedule", "定时分布式调度", DistributeScheduleJob.class, "0/2 * * * * ?"),
                 // 普通调度
-                // ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,7,10,13,16,19 * * ?"),
+                ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,7,10,13,16,19 * * ?"),
                 // 普通调度（20240925不调度11点、14点）
-                ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,7,16,19 * * ?"),
+                // ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,7,16,19 * * ?"),
                 // 普通调度（20240925不调度8点、11点、14点）
                 // ScheduleStrategyImpl.simple("daily_schedule", "普通全表调度", DailyScheduleJob.class, "0 50 1,4,16,19 * * ?"),
                 // 重点表调度

utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/functions/Record2RowDataFunction.java

@@ -11,6 +11,7 @@ import com.lanyuanxiaoyao.service.common.utils.RecordHelper;
 import com.lanyuanxiaoyao.service.common.utils.TableMetaHelper;
 import com.lanyuanxiaoyao.service.sync.configuration.GlobalConfiguration;
 import com.lanyuanxiaoyao.service.sync.functions.type.TypeConverter;
+import com.lanyuanxiaoyao.service.sync.utils.ExceptionUtils;
 import com.lanyuanxiaoyao.service.sync.utils.JacksonUtils;
 import com.lanyuanxiaoyao.service.sync.utils.MetricsUtils;
 import com.lanyuanxiaoyao.service.sync.utils.StatusUtils;
@@ -94,7 +95,7 @@ public class Record2RowDataFunction extends RichMapFunction<Record, List<RowData
         Map<String, Object> current = RecordHelper.getCurrentStatement(record);
         if (Objects.isNull(current)) {
             logger.error("Record: {}", mapper.writeValueAsString(record));
-            throw new RuntimeException("Current cannot be null");
+            return ExceptionUtils.throwAndPrint(logger, "Current cannot be null");
         }
 
         // 如果 update 改变了过滤字段的值也需要先删除

utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/functions/type/TypeConverterV2.java

@@ -5,6 +5,7 @@ import cn.hutool.core.util.StrUtil;
 import com.lanyuanxiaoyao.service.common.Constants;
 import com.lanyuanxiaoyao.service.common.entity.TableMeta;
 import com.lanyuanxiaoyao.service.common.utils.LogHelper;
+import com.lanyuanxiaoyao.service.sync.utils.ExceptionUtils;
 import java.math.BigDecimal;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
@@ -115,7 +116,8 @@ public class TypeConverterV2 implements TypeConverter {
                 return NULLABLE_STRING_SCHEMA;
             }
         } catch (Throwable throwable) {
-            throw new RuntimeException(
+            return ExceptionUtils.throwAndPrint(
+                    logger,
                     StrUtil.format("Convert type failure {} {} {} length: {} scala: {}", table, field, type, length, scala),
                     throwable
             );
@@ -162,7 +164,8 @@ public class TypeConverterV2 implements TypeConverter {
                 return value;
             }
         } catch (Throwable throwable) {
-            throw new RuntimeException(
+            return ExceptionUtils.throwAndPrint(
+                    logger,
                     StrUtil.format("Convert value failure {} {} {}", schema.toString(), name, value),
                     throwable
             );

utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/utils/ExceptionUtils.java

@@ -0,0 +1,26 @@
+package com.lanyuanxiaoyao.service.sync.utils;
+
+import org.slf4j.Logger;
+
+/**
+ * 处理异常抛出和打印
+ *
+ * @author lanyuanxiaoyao
+ * @date 2024-10-22
+ */
+public class ExceptionUtils {
+    public static <T> T throwAndPrint(Logger logger, String content) {
+        logger.error(content);
+        throw new RuntimeException(content);
+    }
+
+    public static <T> T throwAndPrint(Logger logger, Throwable throwable) {
+        logger.error(throwable.getMessage(), throwable);
+        throw new RuntimeException(throwable);
+    }
+
+    public static <T> T throwAndPrint(Logger logger, String content, Throwable throwable) {
+        logger.error(content, throwable);
+        throw new RuntimeException(content, throwable);
+    }
+}

utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/utils/ZkUtils.java

@@ -75,11 +75,9 @@ public class ZkUtils {
                     .withMode(CreateMode.EPHEMERAL)
                     .forPath(lockPath, runMeta.getBytes());
         } catch (KeeperException.NodeExistsException e) {
-            logger.error("Lock exists for " + lockPath, e);
-            throw new RuntimeException(e);
+            ExceptionUtils.throwAndPrint(logger, "Lock exists for " + lockPath, e);
         } catch (Exception e) {
-            logger.error("Unknown error", e);
-            throw new RuntimeException(e);
+            ExceptionUtils.throwAndPrint(logger, "Unknown error", e);
         }
     }
 
@@ -91,8 +89,7 @@ public class ZkUtils {
                 }
             }
         } catch (Exception e) {
-            logger.error("Unknown error", e);
-            throw new RuntimeException(e);
+            ExceptionUtils.throwAndPrint(logger, "Unknown error", e);
         }
     }
 }