Merge branch 'hotfix_1.8_4.0.x_30258' into 1.8_release_4.0.x

Author: gituser

flinkx-kafka/flinkx-kafka-writer/src/main/java/com/dtstack/flinkx/kafka/writer/KafkaOutputFormat.java

@@ -42,6 +42,7 @@ public class KafkaOutputFormat extends KafkaBaseOutputFormat {
 
     @Override
     public void configure(Configuration parameters) {
+        super.configure(parameters);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 86400000);

flinkx-kafka09/flinkx-kafka09-writer/src/main/java/com/dtstack/flinkx/kafka09/writer/Kafka09OutputFormat.java

@@ -18,8 +18,8 @@
 package com.dtstack.flinkx.kafka09.writer;
 
 import com.dtstack.flinkx.kafkabase.Formatter;
+import com.dtstack.flinkx.kafkabase.writer.AddressUtil;
 import com.dtstack.flinkx.kafkabase.writer.KafkaBaseOutputFormat;
-import com.dtstack.flinkx.util.ExceptionUtil;
 import org.apache.flink.configuration.Configuration;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -38,12 +38,12 @@ public class Kafka09OutputFormat extends KafkaBaseOutputFormat {
 
     private String encoding;
     private String brokerList;
-    private transient KafkaProducer<String,String> producer;
+    private transient KafkaProducer<String, String> producer;
     private HeartBeatController heartBeatController;
 
     @Override
     public void configure(Configuration parameters) {
-        props.put("key.serializer",org.apache.kafka.common.serialization.StringSerializer.class.getName());
+        props.put("key.serializer", org.apache.kafka.common.serialization.StringSerializer.class.getName());
         props.put("value.serializer", org.apache.kafka.common.serialization.StringSerializer.class.getName());
         props.put("producer.type", "sync");
         props.put("compression.codec", "none");
@@ -59,17 +59,25 @@ public void configure(Configuration parameters) {
         props.put("metadata.broker.list", brokerList);
         props.put("bootstrap.servers", brokerList);
         producer = new KafkaProducer<>(props);
+
+        LOG.info("brokerList {}", brokerList);
+        String broker = brokerList.split(",")[0];
+        String[] split = broker.split(":");
+
+        if (split.length != 2 || !AddressUtil.telnet(split[0], Integer.parseInt(split[1]))) {
+            throw new RuntimeException("telnet error,brokerList" + brokerList);
+        }
     }
 
     @Override
     protected void emit(Map event) throws IOException {
         heartBeatController.acquire();
         String tp = Formatter.format(event, topic, timezone);
         producer.send(new ProducerRecord<>(tp, event.toString(), objectMapper.writeValueAsString(event)), (metadata, exception) -> {
-            if(Objects.nonNull(exception)){
-                LOG.warn("kafka writeSingleRecordInternal error:{}", exception.getMessage(),exception);
+            if (Objects.nonNull(exception)) {
+                LOG.warn("kafka writeSingleRecordInternal error:{}", exception.getMessage(), exception);
                 heartBeatController.onFailed(exception);
-            }else{
+            } else {
                 heartBeatController.onSuccess();
             }
         });

flinkx-kafka10/flinkx-kafka10-writer/src/main/java/com/dtstack/flinkx/kafka10/writer/Kafka10OutputFormat.java

@@ -41,6 +41,7 @@ public class Kafka10OutputFormat extends KafkaBaseOutputFormat {
 
     @Override
     public void configure(Configuration parameters) {
+        super.configure(parameters);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 86400000);

flinkx-kafka11/flinkx-kafka11-writer/src/main/java/com/dtstack/flinkx/kafka11/writer/Kafka11OutputFormat.java

@@ -41,6 +41,7 @@ public class Kafka11OutputFormat extends KafkaBaseOutputFormat {
 
     @Override
     public void configure(Configuration parameters) {
+        super.configure(parameters);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 86400000);

flinkx-kb/flinkx-kb-writer/src/main/java/com/dtstack/flinkx/kafkabase/writer/AddressUtil.java

@@ -0,0 +1,62 @@
+/*
+ * 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.kafkabase.writer;
+
+import org.apache.commons.net.telnet.TelnetClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+
+/**
+ * @author jiangbo
+ * @date 2020/3/2
+ */
+public class AddressUtil {
+
+    private static Logger logger = LoggerFactory.getLogger(AddressUtil.class);
+
+    public static boolean telnet(String ip,int port){
+        TelnetClient client = null;
+        try{
+            client = new TelnetClient();
+            client.setConnectTimeout(3000);
+            client.connect(ip,port);
+            return true;
+        }catch(Exception e){
+            return false;
+        } finally {
+            try {
+                if (client != null){
+                    client.disconnect();
+                }
+            } catch (Exception e){
+                logger.error("{}",e);
+            }
+        }
+    }
+
+    public static boolean ping(String ip){
+        try{
+            return InetAddress.getByName(ip).isReachable(3000);
+        }catch(Exception e){
+            return false;
+        }
+    }
+}

flinkx-kb/flinkx-kb-writer/src/main/java/com/dtstack/flinkx/kafkabase/writer/KafkaBaseOutputFormat.java

@@ -57,6 +57,16 @@ public class KafkaBaseOutputFormat extends BaseRichOutputFormat {
 
     @Override
     public void configure(Configuration parameters) {
+        if(producerSettings != null && producerSettings.get("bootstrap.servers") != null){
+            String brokerList = producerSettings.get("bootstrap.servers");
+            LOG.info("brokerList->{}",brokerList);
+            String broker = brokerList.split(",")[0];
+            String[] split = broker.split(":");
+
+            if( split.length !=2 ||!AddressUtil.telnet(split[0], Integer.parseInt(split[1]))){
+                throw new RuntimeException("telnet error,brokerList"+brokerList+" please check dataSource is running");
+            }
+        }
     }
 
     @Override