CsvInputFormat: Create a parser per InputEntityReader. (#10923)

RFC4180Parser is not thread safe and cannot be shared across readers.

core/src/main/java/org/apache/druid/data/input/InputEntityReader.java

@@ -28,6 +28,8 @@ import java.io.IOException;
  * InputEntityReader knows how to parse data into {@link InputRow}.
  * This class is <i>stateful</i> and a new InputEntityReader should be created per {@link InputEntity}.
  *
+ * Not thread-safe.
+ *
  * @see IntermediateRowParsingReader
  * @see TextReader
  */

core/src/main/java/org/apache/druid/data/input/impl/CsvInputFormat.java

@@ -38,7 +38,6 @@ import java.util.List;
 public class CsvInputFormat extends FlatTextInputFormat
 {
   private static final char SEPARATOR = ',';
-  private static final RFC4180Parser PARSER = createOpenCsvParser();
 
   @JsonCreator
   public CsvInputFormat(
@@ -68,6 +67,8 @@ public class CsvInputFormat extends FlatTextInputFormat
   @Override
   public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity source, File temporaryDirectory)
   {
+    final RFC4180Parser parser = createOpenCsvParser();
+
     return new DelimitedValueReader(
         inputRowSchema,
         source,
@@ -75,7 +76,7 @@ public class CsvInputFormat extends FlatTextInputFormat
         getColumns(),
         isFindColumnsFromHeader(),
         getSkipHeaderRows(),
-        line -> Arrays.asList(PARSER.parseLine(line))
+        line -> Arrays.asList(parser.parseLine(line))
     );
   }
 

core/src/main/java/org/apache/druid/java/util/common/parsers/Parser.java

@@ -25,6 +25,8 @@ import java.util.Map;
 
 /**
  * Class that can parse Strings into Maps.
+ *
+ * Not thread-safe.
  */
 public interface Parser<K, V>
 {