set new version：v5.0.0-SNAPSHOT

Author: baisui1981

pom.xml

@@ -25,7 +25,7 @@
     <parent>
         <groupId>com.qlangtech.tis</groupId>
         <artifactId>tis</artifactId>
-        <version>4.3.0</version>
+        <version>5.0.0-SNAPSHOT</version>
     </parent>
     <groupId>com.qlangtech.tis.plugins</groupId>
     <artifactId>tis-plugin-parent</artifactId>

pom.xml.versionBackup

@@ -25,7 +25,7 @@
     <parent>
         <groupId>com.qlangtech.tis</groupId>
         <artifactId>tis</artifactId>
-        <version>4.3.0-SNAPSHOT</version>
+        <version>4.3.0</version>
     </parent>
     <groupId>com.qlangtech.tis.plugins</groupId>
     <artifactId>tis-plugin-parent</artifactId>

tis-datax/tis-datax-kingbase-plugin/pom.xml

@@ -38,7 +38,8 @@
         <dependency>
             <groupId>cn.com.kingbase</groupId>
             <artifactId>kingbase8</artifactId>
-            <version>8.6.1</version>
+<!--            <version>8.6.1</version>-->
+            <version>9.0.1</version>
         </dependency>
 
 

tis-incr/tis-flink-cdc-kingbase-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/pglike/FlinkCDCPGLikeSourceFunction.java

@@ -21,34 +21,29 @@
 import com.qlangtech.plugins.incr.flink.cdc.FlinkCol;
 import com.qlangtech.plugins.incr.flink.cdc.SourceChannel;
 import com.qlangtech.plugins.incr.flink.cdc.valconvert.DateTimeConverter;
-import com.qlangtech.plugins.incr.flink.launch.TISFlinkCDCStreamFactory;
 import com.qlangtech.tis.async.message.client.consumer.AsyncMsg;
-import com.qlangtech.tis.async.message.client.consumer.IConsumerHandle;
 import com.qlangtech.tis.async.message.client.consumer.IFlinkColCreator;
 import com.qlangtech.tis.async.message.client.consumer.IMQListener;
 import com.qlangtech.tis.async.message.client.consumer.MQConsumeException;
-import com.qlangtech.tis.coredefine.module.action.TargetResName;
 import com.qlangtech.tis.datax.DataXName;
 import com.qlangtech.tis.datax.IDataxProcessor;
 import com.qlangtech.tis.datax.IDataxReader;
-import com.qlangtech.tis.datax.StoreResourceType;
 import com.qlangtech.tis.plugin.datax.common.BasicDataXRdbmsReader;
 import com.qlangtech.tis.plugin.datax.transformer.RecordTransformerRules;
 import com.qlangtech.tis.plugin.ds.BasicDataSourceFactory;
 import com.qlangtech.tis.plugin.ds.DataSourceFactory.ISchemaSupported;
 import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.plugin.ds.RunningContext;
 import com.qlangtech.tis.plugin.incr.IConsumerRateLimiter;
-import com.qlangtech.tis.plugin.incr.IncrStreamFactory;
 import com.qlangtech.tis.realtime.ReaderSource;
 import com.qlangtech.tis.realtime.dto.DTOStream;
 import com.qlangtech.tis.realtime.transfer.DTO;
 import com.qlangtech.tis.util.IPluginContext;
 import org.apache.commons.lang.StringUtils;
-import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.cdc.connectors.base.source.jdbc.JdbcIncrementalSource;
 import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder.PostgresIncrementalSource;
 
+import java.time.ZoneId;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -99,7 +94,8 @@ public AsyncMsg<List<ReaderSource>> start(IConsumerRateLimiter streamFactory, bo
                          * https://stackoverflow.com/questions/59978213/debezium-could-not-access-file-decoderbufs-using-postgres-11-with-default-plug
                          */
                         debeziumProperties.put("plugin.name", getWALDecoderPluginName());
-                        DateTimeConverter.setDatetimeConverters(KingBaseDateTimeConverter.class.getName(), debeziumProperties);
+                        DateTimeConverter.setDatetimeConverters(
+                                KingBaseDateTimeConverter.class.getName(), debeziumProperties, dsFactory.getTimeZone().map(ZoneId::getId));
 
                         return dbs.getDbStream().map((dbname) -> {
 

tis-incr/tis-flink-cdc-kingbase-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/pglike/KingBaseDateTimeConverter.java

@@ -21,9 +21,7 @@
 import com.qlangtech.plugins.incr.flink.cdc.valconvert.DateTimeConverter;
 
 import java.time.Instant;
-import java.time.LocalDate;
 import java.time.LocalDateTime;
-import java.time.ZoneId;
 
 /**
  * @author: 百岁（baisui@qlangtech.com）
@@ -53,20 +51,23 @@ protected String convertDate(Object input) {
     protected String convertTime(Object input) {
         if (input != null) {
             System.out.println("convertTime:" + input.getClass());
+            throw new UnsupportedOperationException(input.getClass().getName() + ",value:" + input);
+
         }
         return null;
     }
 
     @Override
     protected String convertDateTime(Object input) {
         if (input != null) {
-            System.out.println("convertDateTime:" + input.getClass());
-            throw new UnsupportedOperationException();
+//            System.out.println("convertDateTime:" + input.getClass());
+//            throw new UnsupportedOperationException();
+            return datetimeFormatter.format(LocalDateTime.ofInstant((Instant) input, this.timestampZoneId));
         }
         return null;
     }
 
-    private static final ZoneId zof = ZoneId.of("Z");
+    // private static final ZoneId zof = ZoneId.of("Z");
 
     @Override
     protected String convertTimestamp(Object input) {
@@ -76,7 +77,7 @@ protected String convertTimestamp(Object input) {
 //            }
 
             // System.out.println("timestampZoneId:" + timestampZoneId);
-            return timestampFormatter.format(LocalDateTime.ofInstant((Instant) input, zof));
+            return timestampFormatter.format(LocalDateTime.ofInstant((Instant) input, this.timestampZoneId));
 
 //                System.out.println(">>>>convertTimestamp:" + input.getClass().getName());
 //                System.out.println(">>>>convertTimestamp:" + input.getClass() + ",time:" + timestampFormatter.format((Instant) input));

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/launch/TISFlinkCDCStreamFactory.java

@@ -328,6 +328,8 @@ protected boolean validateAll(IControlMsgHandler msgHandler, Context context, Po
                 }
             }
 
+            plugin.checkUseable(TargetResName.createTargetName(msgHandler.getCollectionName()), true);
+
 
             return super.validateAll(msgHandler, context, postFormVals);
         }

tis-incr/tis-realtime-flink/src/main/java/com/qlangtech/plugins/incr/flink/launch/clustertype/ClusterType.java

@@ -74,6 +74,17 @@ public ServerLaunchToken getLaunchToken(TargetResName collection) {
     }
 
 
+    /**
+     *
+     * @param collection
+     * @param deploying
+     * @param requiredSlots
+     * @return skip flowing check?
+     */
+    protected boolean skipFlowingCheck(TargetResName collection, boolean deploying, int requiredSlots) {
+        return false;
+    }
+
     /**
      *
      * @param collection
@@ -82,8 +93,10 @@ public ServerLaunchToken getLaunchToken(TargetResName collection) {
      */
     public final void checkUseable(TargetResName collection, boolean deploying, int requiredSlots) throws TisException {
 
+        if (this.skipFlowingCheck(collection, deploying, requiredSlots)) {
+            return;
+        }
 
-        // String webInterfaceURL = null;
         try {
             try (ClusterClient restClient = createRestClusterClient()) {
                 //       webInterfaceURL = restClient.getWebInterfaceURL();