日志记录

Author: dapeng

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;
     }