fix kafka cvs pattern error and remove kafka08 monitor

todd5167 · todd5167 · commit 916da8b4e81a · 2018-12-26T16:23:43.000+08:00
diff --git a/kafka08/kafka08-source/pom.xml b/kafka08/kafka08-source/pom.xml
@@ -20,7 +20,7 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-kafka-0.8_2.11</artifactId>
-            <version>1.6.0</version>
+            <version>${flink.version}</version>
         </dependency>
     </dependencies>
 
diff --git a/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -90,19 +90,19 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 			}
 		} else if ("csv".equalsIgnoreCase(kafka08SourceTableInfo.getSourceDataType())) {
 			if (topicIsPattern) {
-				kafkaSrc = new CustomerCsvConsumer(topicName,
+				kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
 						new com.dtstack.flink.sql.source.kafka.deserialization.CustomerCsvDeserialization(typeInformation,
 								kafka08SourceTableInfo.getFieldDelimiter(), kafka08SourceTableInfo.getLengthCheckPolicy()), props);
 			} else {
-				kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
+				kafkaSrc = new CustomerCsvConsumer(topicName,
 						new CustomerCsvDeserialization(typeInformation,
 								kafka08SourceTableInfo.getFieldDelimiter(), kafka08SourceTableInfo.getLengthCheckPolicy()), props);
 			}
 		} else {
 			if (topicIsPattern) {
-				kafkaSrc = new CustomerCommonConsumer(topicName, new com.dtstack.flink.sql.source.kafka.deserialization.CustomerCommonDeserialization(), props);
+				kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new com.dtstack.flink.sql.source.kafka.deserialization.CustomerCommonDeserialization(), props);
 			} else {
-				kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new CustomerCommonDeserialization(), props);
+				kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);
 			}
 		}
 
diff --git a/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/consumer/CustomerCsvConsumer.java b/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/consumer/CustomerCsvConsumer.java
@@ -20,20 +20,11 @@
 
 import com.dtstack.flink.sql.source.AbsDeserialization;
 import com.dtstack.flink.sql.source.kafka.deserialization.CustomerCsvDeserialization;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
-import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
-import org.apache.flink.streaming.connectors.kafka.config.OffsetCommitMode;
-import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
-import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.flink.types.Row;
-import org.apache.flink.util.SerializedValue;
 
 import java.util.Arrays;
-import java.util.Map;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
diff --git a/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/consumer/CustomerJsonConsumer.java b/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/consumer/CustomerJsonConsumer.java
@@ -20,20 +20,11 @@
 
 import com.dtstack.flink.sql.source.AbsDeserialization;
 import com.dtstack.flink.sql.source.kafka.deserialization.CustomerJsonDeserialization;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
-import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer08;
-import org.apache.flink.streaming.connectors.kafka.config.OffsetCommitMode;
-import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
-import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.flink.types.Row;
-import org.apache.flink.util.SerializedValue;
 
 import java.util.Arrays;
-import java.util.Map;
 import java.util.Properties;
 import java.util.regex.Pattern;
 
diff --git a/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/CustomerCommonDeserialization.java b/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/CustomerCommonDeserialization.java
@@ -48,9 +48,9 @@ public class CustomerCommonDeserialization extends AbsDeserialization<Row> imple
 	@Override
 	public Row deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) {
 
-		numInRecord.inc();
-		numInBytes.inc(message.length);
-		numInBytes.inc(messageKey.length);
+		//numInRecord.inc();
+		//numInBytes.inc(message.length);
+		//numInBytes.inc(messageKey.length);
 
 		try {
 			Row row = Row.of(
@@ -63,7 +63,7 @@ public Row deserialize(byte[] messageKey, byte[] message, String topic, int part
 			return row;
 		} catch (Throwable t) {
 			LOG.error(t.getMessage());
-			dirtyDataCounter.inc();
+		//	dirtyDataCounter.inc();
 			return null;
 		}
 	}
diff --git a/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/CustomerCsvDeserialization.java b/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/CustomerCsvDeserialization.java
@@ -80,8 +80,8 @@ public CustomerCsvDeserialization(TypeInformation<Row> typeInfo, String fieldDel
     public Row deserialize(byte[] message) throws IOException {
 
         try {
-            numInRecord.inc();
-            numInBytes.inc(message.length);
+            //numInRecord.inc();
+            //numInBytes.inc(message.length);
             String[] fieldsList = null;
             if (message != null && message.length > 0){
                 fieldsList = new String(message).split(fieldDelimiter);
@@ -103,11 +103,11 @@ public Row deserialize(byte[] message) throws IOException {
                 }
             }
 
-            numInResolveRecord.inc();
+            //numInResolveRecord.inc();
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
-            dirtyDataCounter.inc();
+            //dirtyDataCounter.inc();
             throw new RuntimeException(t);
         }
     }
diff --git a/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/CustomerJsonDeserialization.java b/kafka08/kafka08-source/src/main/java/com/dtstack/flink/sql/source/kafka/deserialization/CustomerJsonDeserialization.java
@@ -80,8 +80,8 @@ public CustomerJsonDeserialization(TypeInformation<Row> typeInfo){
     public Row deserialize(byte[] message) throws IOException {
 
         try {
-            numInRecord.inc();
-            numInBytes.inc(message.length);
+           // numInRecord.inc();
+           // numInBytes.inc(message.length);
 
             JsonNode root = objectMapper.readTree(message);
             Row row = new Row(fieldNames.length);
@@ -102,12 +102,12 @@ public Row deserialize(byte[] message) throws IOException {
                 }
             }
 
-            numInResolveRecord.inc();
+          //  numInResolveRecord.inc();
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
             LOG.error(t.getMessage());
-            dirtyDataCounter.inc();
+           // dirtyDataCounter.inc();
             return null;
         }
     }
diff --git a/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -100,19 +100,19 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
             }
         } else if ("csv".equalsIgnoreCase(kafka09SourceTableInfo.getSourceDataType())) {
             if (topicIsPattern) {
-                kafkaSrc = new CustomerCsvConsumer(topicName,
+                kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
                         new CustomerCsvDeserialization(typeInformation,
                                 kafka09SourceTableInfo.getFieldDelimiter(), kafka09SourceTableInfo.getLengthCheckPolicy()), props);
             } else {
-                kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
+                kafkaSrc = new CustomerCsvConsumer(topicName,
                         new CustomerCsvDeserialization(typeInformation,
                                 kafka09SourceTableInfo.getFieldDelimiter(), kafka09SourceTableInfo.getLengthCheckPolicy()), props);
             }
         } else {
             if (topicIsPattern) {
-                kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);
-            } else {
                 kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new CustomerCommonDeserialization(), props);
+            } else {
+                kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);
             }
         }
 
diff --git a/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -103,19 +103,19 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 			}
 		} else if ("csv".equalsIgnoreCase(kafka010SourceTableInfo.getSourceDataType())) {
 			if (topicIsPattern) {
-				kafkaSrc = new CustomerCsvConsumer(topicName,
+				kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
 						new CustomerCsvDeserialization(typeInformation,
 								kafka010SourceTableInfo.getFieldDelimiter(), kafka010SourceTableInfo.getLengthCheckPolicy()), props);
 			} else {
-				kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
+				kafkaSrc = new CustomerCsvConsumer(topicName,
 						new CustomerCsvDeserialization(typeInformation,
 								kafka010SourceTableInfo.getFieldDelimiter(), kafka010SourceTableInfo.getLengthCheckPolicy()), props);
 			}
 		} else {
 			if (topicIsPattern) {
-				kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);
-			} else {
 				kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new CustomerCommonDeserialization(), props);
+			} else {
+				kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);
 			}
 		}
 
diff --git a/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -100,19 +100,19 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 			}
 		} else if ("csv".equalsIgnoreCase(kafka011SourceTableInfo.getSourceDataType())) {
 			if (topicIsPattern) {
-				kafkaSrc = new CustomerCsvConsumer(topicName,
+				kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
 						new CustomerCsvDeserialization(typeInformation,
 								kafka011SourceTableInfo.getFieldDelimiter(), kafka011SourceTableInfo.getLengthCheckPolicy()), props);
 			} else {
-				kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),
+				kafkaSrc = new CustomerCsvConsumer(topicName,
 						new CustomerCsvDeserialization(typeInformation,
 								kafka011SourceTableInfo.getFieldDelimiter(), kafka011SourceTableInfo.getLengthCheckPolicy()), props);
 			}
 		} else {
 			if (topicIsPattern) {
-				kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);
-			} else {
 				kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new CustomerCommonDeserialization(), props);
+			} else {
+				kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);
 			}
 		}
 

Original file line number	Diff line number	Diff line change
`@@ -90,19 +90,19 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv`
`90`	`90`	`}`
`91`	`91`	`} else if ("csv".equalsIgnoreCase(kafka08SourceTableInfo.getSourceDataType())) {`
`92`	`92`	`if (topicIsPattern) {`
`93`		`- kafkaSrc = new CustomerCsvConsumer(topicName,`
	`93`	`+ kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),`
`94`	`94`	`new com.dtstack.flink.sql.source.kafka.deserialization.CustomerCsvDeserialization(typeInformation,`
`95`	`95`	`kafka08SourceTableInfo.getFieldDelimiter(), kafka08SourceTableInfo.getLengthCheckPolicy()), props);`
`96`	`96`	`} else {`
`97`		`- kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),`
	`97`	`+ kafkaSrc = new CustomerCsvConsumer(topicName,`
`98`	`98`	`new CustomerCsvDeserialization(typeInformation,`
`99`	`99`	`kafka08SourceTableInfo.getFieldDelimiter(), kafka08SourceTableInfo.getLengthCheckPolicy()), props);`
`100`	`100`	`}`
`101`	`101`	`} else {`
`102`	`102`	`if (topicIsPattern) {`
`103`		`- kafkaSrc = new CustomerCommonConsumer(topicName, new com.dtstack.flink.sql.source.kafka.deserialization.CustomerCommonDeserialization(), props);`
	`103`	`+ kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new com.dtstack.flink.sql.source.kafka.deserialization.CustomerCommonDeserialization(), props);`
`104`	`104`	`} else {`
`105`		`- kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new CustomerCommonDeserialization(), props);`
	`105`	`+ kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);`
`106`	`106`	`}`
`107`	`107`	`}`
`108`	`108`
Original file line number	Diff line number	Diff line change
`@@ -80,8 +80,8 @@ public CustomerCsvDeserialization(TypeInformation<Row> typeInfo, String fieldDel`
`80`	`80`	`public Row deserialize(byte[] message) throws IOException {`
`81`	`81`
`82`	`82`	`try {`
`83`		`- numInRecord.inc();`
`84`		`- numInBytes.inc(message.length);`
	`83`	`+ //numInRecord.inc();`
	`84`	`+ //numInBytes.inc(message.length);`
`85`	`85`	`String[] fieldsList = null;`
`86`	`86`	`if (message != null && message.length > 0){`
`87`	`87`	`fieldsList = new String(message).split(fieldDelimiter);`
`@@ -103,11 +103,11 @@ public Row deserialize(byte[] message) throws IOException {`
`103`	`103`	`}`
`104`	`104`	`}`
`105`	`105`
`106`		`- numInResolveRecord.inc();`
	`106`	`+ //numInResolveRecord.inc();`
`107`	`107`	`return row;`
`108`	`108`	`} catch (Throwable t) {`
`109`	`109`	`//add metric of dirty data`
`110`		`- dirtyDataCounter.inc();`
	`110`	`+ //dirtyDataCounter.inc();`
`111`	`111`	`throw new RuntimeException(t);`
`112`	`112`	`}`
`113`	`113`	`}`
Original file line number	Diff line number	Diff line change
`@@ -80,8 +80,8 @@ public CustomerJsonDeserialization(TypeInformation<Row> typeInfo){`
`80`	`80`	`public Row deserialize(byte[] message) throws IOException {`
`81`	`81`
`82`	`82`	`try {`
`83`		`- numInRecord.inc();`
`84`		`- numInBytes.inc(message.length);`
	`83`	`+ // numInRecord.inc();`
	`84`	`+ // numInBytes.inc(message.length);`
`85`	`85`
`86`	`86`	`JsonNode root = objectMapper.readTree(message);`
`87`	`87`	`Row row = new Row(fieldNames.length);`
`@@ -102,12 +102,12 @@ public Row deserialize(byte[] message) throws IOException {`
`102`	`102`	`}`
`103`	`103`	`}`
`104`	`104`
`105`		`- numInResolveRecord.inc();`
	`105`	`+ // numInResolveRecord.inc();`
`106`	`106`	`return row;`
`107`	`107`	`} catch (Throwable t) {`
`108`	`108`	`//add metric of dirty data`
`109`	`109`	`LOG.error(t.getMessage());`
`110`		`- dirtyDataCounter.inc();`
	`110`	`+ // dirtyDataCounter.inc();`
`111`	`111`	`return null;`
`112`	`112`	`}`
`113`	`113`	`}`
Original file line number	Diff line number	Diff line change
`@@ -100,19 +100,19 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv`
`100`	`100`	`}`
`101`	`101`	`} else if ("csv".equalsIgnoreCase(kafka09SourceTableInfo.getSourceDataType())) {`
`102`	`102`	`if (topicIsPattern) {`
`103`		`- kafkaSrc = new CustomerCsvConsumer(topicName,`
	`103`	`+ kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),`
`104`	`104`	`new CustomerCsvDeserialization(typeInformation,`
`105`	`105`	`kafka09SourceTableInfo.getFieldDelimiter(), kafka09SourceTableInfo.getLengthCheckPolicy()), props);`
`106`	`106`	`} else {`
`107`		`- kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),`
	`107`	`+ kafkaSrc = new CustomerCsvConsumer(topicName,`
`108`	`108`	`new CustomerCsvDeserialization(typeInformation,`
`109`	`109`	`kafka09SourceTableInfo.getFieldDelimiter(), kafka09SourceTableInfo.getLengthCheckPolicy()), props);`
`110`	`110`	`}`
`111`	`111`	`} else {`
`112`	`112`	`if (topicIsPattern) {`
`113`		`- kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);`
`114`		`- } else {`
`115`	`113`	`kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new CustomerCommonDeserialization(), props);`
	`114`	`+ } else {`
	`115`	`+ kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);`
`116`	`116`	`}`
`117`	`117`	`}`
`118`	`118`
Original file line number	Diff line number	Diff line change
`@@ -103,19 +103,19 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv`
`103`	`103`	`}`
`104`	`104`	`} else if ("csv".equalsIgnoreCase(kafka010SourceTableInfo.getSourceDataType())) {`
`105`	`105`	`if (topicIsPattern) {`
`106`		`- kafkaSrc = new CustomerCsvConsumer(topicName,`
	`106`	`+ kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),`
`107`	`107`	`new CustomerCsvDeserialization(typeInformation,`
`108`	`108`	`kafka010SourceTableInfo.getFieldDelimiter(), kafka010SourceTableInfo.getLengthCheckPolicy()), props);`
`109`	`109`	`} else {`
`110`		`- kafkaSrc = new CustomerCsvConsumer(Pattern.compile(topicName),`
	`110`	`+ kafkaSrc = new CustomerCsvConsumer(topicName,`
`111`	`111`	`new CustomerCsvDeserialization(typeInformation,`
`112`	`112`	`kafka010SourceTableInfo.getFieldDelimiter(), kafka010SourceTableInfo.getLengthCheckPolicy()), props);`
`113`	`113`	`}`
`114`	`114`	`} else {`
`115`	`115`	`if (topicIsPattern) {`
`116`		`- kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);`
`117`		`- } else {`
`118`	`116`	`kafkaSrc = new CustomerCommonConsumer(Pattern.compile(topicName), new CustomerCommonDeserialization(), props);`
	`117`	`+ } else {`
	`118`	`+ kafkaSrc = new CustomerCommonConsumer(topicName, new CustomerCommonDeserialization(), props);`
`119`	`119`	`}`
`120`	`120`	`}`
`121`	`121`