fix bug for data preview before sync trigger ,issue: datavane/tis#476

Author: baisui1981

tis-datax/executor/tis-datax-executor/src/main/java/com/qlangtech/tis/datax/common/DataXRealExecutor.java

@@ -157,7 +157,7 @@ public PreviewRecords previewRecords(String tableName, QueryCriteria queryCriter
             Optional<Pair<String, List<String>>> transformer
                     = transformerRules.map((trule) -> Pair.of(tableName, trule.relevantColKeys()));
 
-            this.startPipeline(readerCfg, transformer, (jobContainer) -> {
+            this.startPipeline(tableName,readerCfg, transformer, (jobContainer) -> {
                 jobContainer.setAttr(ThreadLocalRows.class, rows);
             });
 
@@ -177,7 +177,7 @@ public PreviewRecords previewRecords(String tableName, QueryCriteria queryCriter
      * @param transformer
      * @param jobContainerSetter
      */
-    public void startPipeline(Configuration readerCfg, Optional<Pair<String, List<String>>> transformer, Consumer<JobContainer> jobContainerSetter) {
+    public void startPipeline(String tableName,Configuration readerCfg, Optional<Pair<String, List<String>>> transformer, Consumer<JobContainer> jobContainerSetter) {
         Objects.requireNonNull(readerCfg);
         PerfTrace.getInstance(false, -1111, -1111, 0, false);
         Configuration allConf = IOUtils.loadResourceFromClasspath(DataxExecutor.class //
@@ -197,7 +197,7 @@ public void startPipeline(Configuration readerCfg, Optional<Pair<String, List<St
 
 
                     Configuration c = Configuration.newDefault();
-                    c.set(CoreConstant.JOB_TRANSFORMER_NAME, transformer.map((p) -> p.getKey()).orElse("testTab"));
+                    c.set(CoreConstant.JOB_TRANSFORMER_NAME, tableName);
                     transformer.ifPresent((tt) -> {
                         Pair<String, List<String>> t = tt;
                         c.set(CoreConstant.JOB_TRANSFORMER_RELEVANT_KEYS, t.getRight());

tis-datax/executor/tis-datax-executor/src/main/java/com/qlangtech/tis/datax/common/WriterPluginMeta.java

@@ -51,12 +51,12 @@ public WriterPluginMeta(String pluginKey, String streamwriterClass, Configuratio
         this.conf = conf;
     }
 
-    public static void realExecute(final String dataXName, final Configuration readerCfg
+    public static void realExecute(final String dataXName,String tableName ,final Configuration readerCfg
                                    // , IDataXPluginMeta dataxReader
             , WriterPluginMeta writerPluginMeta
             , Optional<Pair<String, List<String>>> transformer, Optional<JarLoader> jarLoader) throws IllegalAccessException {
         realExecute(dataXName, writerPluginMeta, jarLoader)
-                .startPipeline(readerCfg, transformer, (jobContainer) -> {
+                .startPipeline(tableName,readerCfg, transformer, (jobContainer) -> {
                 });
 
     }

tis-datax/tis-datax-common-plugin/src/main/java/com/qlangtech/tis/plugin/datax/writer/DataGridWriter.java

@@ -25,9 +25,7 @@
 import com.alibaba.datax.common.spi.Writer;
 import com.alibaba.datax.common.util.Configuration;
 import com.google.common.collect.Lists;
-import org.apache.commons.lang3.Range;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 
@@ -81,7 +79,8 @@ public void startWrite(RecordReceiver lineReceiver) {
                 // gridRows.add(record);
                 //recordToString(record);
                 if (++readCount > queryCriteria.getPageSize()) {
-                    throw new IllegalStateException("readCount:" + readCount + " can not more than page size:" + queryCriteria.getPageSize());
+                    // throw new IllegalStateException("readCount:" + readCount + " can not more than page size:" + queryCriteria.getPageSize());
+                    return;
                 }
             }
         }

tis-datax/tis-datax-local-executor/src/main/java/com/qlangtech/tis/plugin/datax/DataXPipelinePreviewProcessorExecutor.java

@@ -27,7 +27,6 @@
 import com.qlangtech.tis.datax.DataXJobSingleProcessorExecutor;
 import com.qlangtech.tis.datax.DataXJobSubmit.InstanceType;
 import com.qlangtech.tis.datax.DataXName;
-import com.qlangtech.tis.datax.DataxPrePostConsumer;
 import com.qlangtech.tis.datax.IDataXTaskRelevant;
 import com.qlangtech.tis.datax.TimeFormat;
 import com.qlangtech.tis.datax.preview.IPreviewRowsDataService;
@@ -42,6 +41,7 @@
 import com.qlangtech.tis.rpc.grpc.datax.preview.PreviewRowsDataCriteria.Builder;
 import com.qlangtech.tis.rpc.grpc.datax.preview.PreviewRowsDataResponse;
 import com.qlangtech.tis.rpc.grpc.datax.preview.StringValue;
+import com.qlangtech.tis.web.start.TisAppLaunch;
 import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
@@ -260,7 +260,7 @@ protected String getMainClassName() {
 
     @Override
     protected File getWorkingDirectory() {
-        return DataXJobInfo.getDataXExecutorDir();
+        return TisAppLaunch.isTestMock() ? new File("/opt/tis/tis-datax-executor") : DataXJobInfo.getDataXExecutorDir();
     }
 
     @Override

tis-datax/tis-datax-test-common/src/main/java/com/qlangtech/tis/plugin/common/ReaderTemplate.java

@@ -159,7 +159,7 @@ public static void realExecute(final String dataXName, final Configuration reade
                 //"        \"print\": true\n" +
                 "        \"" + Key.PATH + "\": \"" + writeFile.getParentFile().getAbsolutePath() + "\",\n" //
                 + "        \"" + Key.FILE_NAME + "\": \"" + writeFile.getName() + "\"\n" + "    }\n" + "}"));
-        WriterPluginMeta.realExecute(dataXName, readerCfg, writerPluginMeta, transformer, Optional.empty());
+        WriterPluginMeta.realExecute(dataXName, "testTabName", readerCfg, writerPluginMeta, transformer, Optional.empty());
 
 //        Objects.requireNonNull(readerCfg);
 //        final JarLoader uberClassLoader = new JarLoader(new String[]{"."});

tis-incr/tis-flink-cdc-common/src/main/java/com/qlangtech/plugins/incr/flink/cdc/pglike/FlinkCDCPGLikeSourceFactory.java

@@ -33,6 +33,7 @@
 import com.qlangtech.tis.plugin.ds.ISelectedTab;
 import com.qlangtech.tis.runtime.module.misc.IControlMsgHandler;
 import com.qlangtech.tis.runtime.module.misc.IFieldErrorHandler;
+import com.qlangtech.tis.trigger.util.UnCacheString;
 import com.qlangtech.tis.util.IPluginContext;
 import io.debezium.config.Field;
 import org.apache.commons.lang3.tuple.Pair;
@@ -113,7 +114,7 @@ public BasePGLikeDescriptor() {
             this.debeziumProps = Lists.newArrayList(
                     Pair.of((opts) -> {
                                 opts.add(FIELD_KEY_SLOT_NAME, getSoltNameField()
-                                        , new OverwriteProps().setDftVal(IPluginContext.getThreadLocalInstance().getCollectionName().getPipelineName()));
+                                        , new OverwriteProps().setDftVal(new UnCacheString<>(() -> IPluginContext.getThreadLocalInstance().getCollectionName().getPipelineName())));
                             }
                             , (debeziumProperties, sourceFactory) -> {
                                 debeziumProperties.setProperty(getSoltNameField().name(), sourceFactory.slotName);