修改kafka连接

src/main/java/com/dsk/flink/dsc/common/sink/MysqlDataTransferSinkBatch.java

@@ -30,7 +30,7 @@ public class MysqlDataTransferSinkBatch extends RichSinkFunction<String> {
     EnvProperties envProps;
     private static transient ExecutorService executorService;
     private static transient DruidDataSource dataSource;
-    private static final int BATCH_SIZE = 500;
+    private static final int BATCH_SIZE = 100;
     private static final int FLUSH_INTERVAL = 500;
     private static final int INIT_CAPACITY = (int)(BATCH_SIZE / 0.75 + 1);
     private ArrayBlockingQueue<String> sqlBatch = new ArrayBlockingQueue<>(BATCH_SIZE * 2);

src/main/java/com/dsk/flink/dsc/sync/SyncCustomerDataSource.java

@@ -108,7 +108,7 @@ public class SyncCustomerDataSource {
                                         Collector<String> out) throws Exception {
                         Tuple3<String, String, Long> maxTsElement = null;
                         for (Tuple3<String, String, Long> element : elements) {
-                            if (maxTsElement == null || element.f2 > maxTsElement.f2) {
+                            if (maxTsElement == null || element.f2 >= maxTsElement.f2) {
                                 maxTsElement = element;
                             }
                         }

src/main/java/com/dsk/flink/dsc/utils/EtlUtils.java

@@ -15,7 +15,7 @@ public class EtlUtils {
         properties.setProperty("auto.offset.reset", "earliest");
         properties.setProperty("sasl.jaas.config", getSaslJaasConfig(username,password));
         properties.setProperty("security.protocol", "SASL_PLAINTEXT");
-        properties.setProperty("sasl.mechanism", "SCRAM-SHA-256");
+        properties.setProperty("sasl.mechanism", "SCRAM-SHA-512");
         properties.setProperty("fetch.max.bytes", "1048576"); //1M
         properties.setProperty("flink.consumer.max.fetch.size", "524288");//512k
         properties.setProperty("session.timeout.ms", "30000");