[hotfix-794][core] fix job Configuration item is missing NPE

flinkx-core/src/main/java/com/dtstack/flinkx/Main.java

@@ -21,6 +21,7 @@
 import com.dtstack.flinkx.cdc.RestorationFlatMap;
 import com.dtstack.flinkx.cdc.monitor.fetch.FetcherBase;
 import com.dtstack.flinkx.cdc.monitor.store.StoreBase;
+import com.dtstack.flinkx.conf.OperatorConf;
 import com.dtstack.flinkx.conf.SpeedConf;
 import com.dtstack.flinkx.conf.SyncConf;
 import com.dtstack.flinkx.constants.ConstantValue;
@@ -37,6 +38,7 @@
 import com.dtstack.flinkx.source.SourceFactory;
 import com.dtstack.flinkx.sql.parser.SqlParser;
 import com.dtstack.flinkx.throwable.FlinkxRuntimeException;
+import com.dtstack.flinkx.throwable.JobConfigException;
 import com.dtstack.flinkx.util.DataSyncFactoryUtil;
 import com.dtstack.flinkx.util.ExecuteProcessHelper;
 import com.dtstack.flinkx.util.FactoryHelper;
@@ -243,6 +245,8 @@ private static DataStream<RowData> syncStreamToTable(
         Table sourceTable =
                 tableEnv.fromDataStream(
                         sourceDataStream, expressionList.toArray(new Expression[0]));
+
+        checkTableConf(config.getReader());
         tableEnv.createTemporaryView(config.getReader().getTable().getTableName(), sourceTable);
 
         String transformSql = config.getJob().getTransformer().getTransformSql();
@@ -254,6 +258,8 @@ private static DataStream<RowData> syncStreamToTable(
                 TableUtil.getTypeInformation(tableDataTypes, tableFieldNames);
         DataStream<RowData> dataStream =
                 tableEnv.toRetractStream(adaptTable, typeInformation).map(f -> f.f1);
+
+        checkTableConf(config.getWriter());
         tableEnv.createTemporaryView(config.getWriter().getTable().getTableName(), dataStream);
 
         return dataStream;
@@ -331,4 +337,18 @@ private static void configStreamExecutionEnvironment(
         }
         PluginUtil.registerShipfileToCachedFile(options.getAddShipfile(), env);
     }
+
+    /**
+     * Check required config item.
+     *
+     * @param operatorConf
+     */
+    private static void checkTableConf(OperatorConf operatorConf) {
+        if (operatorConf.getTable() == null) {
+            throw new JobConfigException(operatorConf.getName(), "table", "is missing");
+        }
+        if (StringUtils.isEmpty(operatorConf.getTable().getTableName())) {
+            throw new JobConfigException(operatorConf.getName(), "table.tableName", "is missing");
+        }
+    }
 }

flinkx-core/src/main/java/com/dtstack/flinkx/throwable/JobConfigException.java

@@ -0,0 +1,47 @@
+/*
+ * 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 com.dtstack.flinkx.throwable;
+
+/**
+ * @author zhujinming
+ * @since 2022/5/12 4:50 下午
+ */
+public class JobConfigException extends FlinkxRuntimeException {
+
+    private static final long serialVersionUID = 140341232132169L;
+
+    public JobConfigException(String connector, String configItem, String message) {
+        super(
+                String.format(
+                        "Job connector: [%s] configItem: [%s], %s",
+                        connector, configItem, message));
+    }
+
+    public JobConfigException(Throwable cause) {
+        super(cause);
+    }
+
+    public JobConfigException(
+            String connector, String configItem, String message, Throwable cause) {
+        super(
+                String.format(
+                        "Job connector: [%s] configItem: [%s], %s", connector, configItem, message),
+                cause);
+    }
+}