CSV 파서의 컬럼 개수 검증 로직 개선

Author: fharenheit

nifi-custom-controllers/src/main/java/io/datadynamics/nifi/record/csv/AbstractCSVRecordReader.java

@@ -13,21 +13,30 @@
 abstract public class AbstractCSVRecordReader implements RecordReader {
 
     protected final ComponentLog logger;
+
     protected final boolean hasHeader;
+
     protected final boolean ignoreHeader;
+
     private final boolean trimDoubleQuote;
 
     protected final Supplier<DateFormat> LAZY_DATE_FORMAT;
+
     protected final Supplier<DateFormat> LAZY_TIME_FORMAT;
+
     protected final Supplier<DateFormat> LAZY_TIMESTAMP_FORMAT;
 
     protected final String dateFormat;
+
     protected final String timeFormat;
+
     protected final String timestampFormat;
 
     protected final RecordSchema schema;
-    private final Integer fieldCount;
-    private final boolean failOnMismatchFieldCount;
+
+    final Integer fieldCount;
+
+    final boolean failOnMismatchFieldCount;
 
     AbstractCSVRecordReader(final ComponentLog logger, final RecordSchema schema, final boolean hasHeader, final boolean ignoreHeader,
                             final String dateFormat, final String timeFormat, final String timestampFormat, final boolean trimDoubleQuote) {

nifi-custom-controllers/src/main/java/io/datadynamics/nifi/record/csv/CSVRecordReader.java

@@ -3,6 +3,7 @@
 import org.apache.commons.io.input.BOMInputStream;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.serialization.MalformedRecordException;
+import org.apache.nifi.serialization.SchemaValidationException;
 import org.apache.nifi.serialization.record.Record;
 import org.apache.nifi.serialization.record.*;
 import shaded.org.apache.commons.csv.CSVFormat;
@@ -39,7 +40,7 @@ public CSVRecordReader(final InputStream in, final ComponentLog logger, final Re
             withHeader = csvFormat.withHeader(schema.getFieldNames().toArray(new String[0]));
         }
 
-        withHeader = csvFormat.withFieldCount(fieldCount).withValidateFieldCount(failOnMismatchFieldCount);
+        withHeader = withHeader.withFieldCount(fieldCount).withValidateFieldCount(failOnMismatchFieldCount);
 
         csvParser = new CSVParser(reader, withHeader);
     }
@@ -61,6 +62,10 @@ public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFie
                 for (int i = 0; i < csvRecord.size(); i++) {
                     final String rawValue = csvRecord.get(i);
 
+                    if (failOnMismatchFieldCount && fieldCount != csvRecord.size()) {
+                        throw new SchemaValidationException(String.format("CSV 파일의 컬럼 개수와 파싱한 컬럼 개수가 상이합니다. 원래 컬럼 개수 : %s, 파싱한 컬럼 개수 : %s", fieldCount, csvRecord.size()));
+                    }
+
                     final String rawFieldName;
                     final DataType dataType;
                     if (i >= numFieldNames) {