Merge branch 'feat_1.8_keyPartition_v2' into 'feat_keyPartition_mergeTest'

Feat 1.8 key partition v2



See merge request !259

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java

@@ -36,6 +36,8 @@
  */
 public class SerializationMetricWrapper implements SerializationSchema<Row> {
 
+    private static final long serialVersionUID = 1L;
+
     private SerializationSchema<Row> serializationSchema;
 
     private transient RuntimeContext runtimeContext;

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerFlinkPartition.java

@@ -1,8 +1,6 @@
 package com.dtstack.flink.sql.sink.kafka;
 
 
-import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
-
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.util.Preconditions;
 

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/CustomerKeyedSerializationSchema.java

@@ -11,10 +11,14 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.atomic.AtomicLong;
+
 public class CustomerKeyedSerializationSchema implements KeyedSerializationSchema<Row> {
 
     private static final Logger LOG = LoggerFactory.getLogger(CustomerKeyedSerializationSchema.class);
 
+    private static final AtomicLong COUNTER = new AtomicLong(0L);
+
     private static final long serialVersionUID = 1L;
     private final SerializationMetricWrapper serializationMetricWrapper;
     private String[] partitionKeys;
@@ -57,7 +61,9 @@ private byte[] serializeJsonKey(JsonRowSerializationSchema jsonRowSerializationS
             }
             return sb.toString().getBytes();
         } catch (Exception e){
-            LOG.error("serializeJsonKey error", e);
+            if(COUNTER.getAndIncrement() % 1000 == 0){
+                LOG.error("serializeJsonKey error", e);
+            }
         }
         return null;
     }

kafka/kafka-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer.java

@@ -24,6 +24,8 @@
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Optional;
 import java.util.Properties;
@@ -37,6 +39,10 @@
  */
 public class KafkaProducer extends FlinkKafkaProducer<Row> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaProducer.class);
+
+    private static final long serialVersionUID = 1L;
+
     private SerializationMetricWrapper serializationMetricWrapper;
 
     public KafkaProducer(String topicId, SerializationSchema<Row> serializationSchema, Properties producerConfig, Optional<FlinkKafkaPartitioner<Row>> customPartitioner, String[] parititonKeys) {
@@ -46,6 +52,7 @@ public KafkaProducer(String topicId, SerializationSchema<Row> serializationSchem
 
     @Override
     public void open(Configuration configuration) throws Exception {
+        LOG.warn("---open KafkaProducer--");
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
         serializationMetricWrapper.initMetric();

kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer09.java

@@ -24,6 +24,8 @@
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer09;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Optional;
 import java.util.Properties;
@@ -37,6 +39,10 @@
  */
 public class KafkaProducer09 extends FlinkKafkaProducer09<Row> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaProducer09.class);
+
+    private static final long serialVersionUID = 1L;
+
     private SerializationMetricWrapper serializationMetricWrapper;
 
     public KafkaProducer09(String topicId, SerializationSchema<Row> serializationSchema, Properties producerConfig, Optional<FlinkKafkaPartitioner<Row>> customPartitioner,String[] partitionKeys) {
@@ -46,6 +52,7 @@ public KafkaProducer09(String topicId, SerializationSchema<Row> serializationSch
 
     @Override
     public void open(Configuration configuration) {
+        LOG.info("----KafkaProducer09 open ---");
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
         serializationMetricWrapper.initMetric();

kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer010.java

@@ -24,6 +24,8 @@
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Optional;
 import java.util.Properties;
@@ -37,6 +39,10 @@
  */
 public class KafkaProducer010 extends FlinkKafkaProducer010<Row> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaProducer010.class);
+
+    private static final long serialVersionUID = 1L;
+
     private SerializationMetricWrapper serializationMetricWrapper;
 
     public KafkaProducer010(String topicId, SerializationSchema<Row> serializationSchema, Properties producerConfig, Optional<FlinkKafkaPartitioner<Row>> customPartitioner, String[] partitionKeys) {
@@ -46,6 +52,7 @@ public KafkaProducer010(String topicId, SerializationSchema<Row> serializationSc
 
     @Override
     public void open(Configuration configuration) {
+        LOG.info("----open KafkaProducer010 --");
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
         serializationMetricWrapper.initMetric();

kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -108,8 +108,8 @@ public TypeInformation<Row> getRecordType() {
     @Override
     public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 
-        RichSinkFunction<Row> kafkaProducer010 =  new KafkaProducer010Factory().createKafkaProducer(kafka10SinkTableInfo, getOutputType().getTypeAt(1), properties,
-                Optional.of(new CustomerFlinkPartition<Row>()), partitionKeys);
+        RichSinkFunction<Row> kafkaProducer010 = new KafkaProducer010Factory().createKafkaProducer(kafka10SinkTableInfo, getOutputType().getTypeAt(1), properties,
+                Optional.of(new CustomerFlinkPartition<>()), partitionKeys);
 
         DataStream<Row> mapDataStream = dataStream.filter((Tuple2<Boolean, Row> record) -> record.f0)
                 .map((Tuple2<Boolean, Row> record) -> record.f1)

kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaProducer011.java

@@ -25,6 +25,8 @@
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
 import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Optional;
 import java.util.Properties;
@@ -38,6 +40,10 @@
  */
 public class KafkaProducer011 extends FlinkKafkaProducer011<Row> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KafkaProducer011.class);
+
+    private static final long serialVersionUID = 1L;
+
     private SerializationMetricWrapper serializationMetricWrapper;
 
     public KafkaProducer011(String topicId, SerializationSchema<Row> serializationSchema, Properties producerConfig, Optional<FlinkKafkaPartitioner<Row>> customPartitioner, String[] partitionKeys) {
@@ -47,6 +53,7 @@ public KafkaProducer011(String topicId, SerializationSchema<Row> serializationSc
 
     @Override
     public void open(Configuration configuration) throws Exception {
+        LOG.info("--KafkaProducer011 open--");
         RuntimeContext runtimeContext = getRuntimeContext();
         serializationMetricWrapper.setRuntimeContext(runtimeContext);
         serializationMetricWrapper.initMetric();