CSV connector's task hanged in infinite loop after meeting record it can't process

[MaxSamsonov]

Hello!

We're testing SpoolDirCsvSourceConnector and faced following condition:

1. Part of file was processed (2000 rows)

2. Error from log:
   [2021-08-24 18:55:46,547] ERROR Exception encountered processing line 5692 of /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv. (com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask:265)
   java.io.IOException: Unterminated quoted field at end of CSV line. Beginning of lost text: [;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
   ]
   at com.opencsv.CSVReader.readNext(CSVReader.java:353)
   at com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceTask.process(SpoolDirCsvSourceTask.java:105)
   at com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask.read(AbstractSourceTask.java:254)
   at com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask.poll(AbstractSourceTask.java:148)
   at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:289)
   at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:256)
   at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
   at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
   at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
   at java.base/java.lang.Thread.run(Thread.java:829)
   [2021-08-24 18:55:46,549] INFO Closing /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv (com.github.jcustenborder.kafka.connect.spooldir.InputFile:173)
   [2021-08-24 18:55:46,550] INFO Removing processing flag /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv.PROCESSING (com.github.jcustenborder.kafka.connect.spooldir.InputFile:177)
   [2021-08-24 18:55:46,551] ERROR Error during processing, moving /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv to /connect_integrations/848_quality/DEV/SWAP-ERROR. (com.github.jcustenborder.kafka.connect.spooldir.AbstractCleanUpPolicy:90)
   [2021-08-24 18:55:46,552] INFO Moving /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv to /connect_integrations/848_quality/DEV/SWAP-ERROR/SWAP-2021_06_0001_0.csv (com.github.jcustenborder.kafka.connect.spooldir.InputFile:204)

3. File successfully moved to error folder.

4. No new files can be processed, connector task seems to hang on processing line with error on every "empty.poll.wait.ms" interval

[2021-08-24 19:29:54,234] ERROR Exception encountered processing line 5692 of /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv. (com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask:265)
at com.opencsv.CSVReader.readNext(CSVReader.java:353)
[2021-08-24 19:29:54,234] INFO Closing /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv (com.github.jcustenborder.kafka.connect.spooldir.InputFile:173)
[2021-08-24 19:29:54,236] ERROR Error during processing, moving /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv to /connect_integrations/848_quality/DEV/SWAP-ERROR. (com.github.jcustenborder.kafka.connect.spooldir.AbstractCleanUpPolicy:90)
[2021-08-24 19:30:00,236] ERROR Exception encountered processing line 5692 of /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv. (com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask:265)
at com.opencsv.CSVReader.readNext(CSVReader.java:353)
[2021-08-24 19:30:00,237] INFO Closing /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv (com.github.jcustenborder.kafka.connect.spooldir.InputFile:173)
[2021-08-24 19:30:00,239] ERROR Error during processing, moving /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv to /connect_integrations/848_quality/DEV/SWAP-ERROR. (com.github.jcustenborder.kafka.connect.spooldir.AbstractCleanUpPolicy:90)
[2021-08-24 19:30:06,240] ERROR Exception encountered processing line 5692 of /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv. (com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask:265)
at com.opencsv.CSVReader.readNext(CSVReader.java:353)
[2021-08-24 19:30:06,240] INFO Closing /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv (com.github.jcustenborder.kafka.connect.spooldir.InputFile:173)
[2021-08-24 19:30:06,242] ERROR Error during processing, moving /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv to /connect_integrations/848_quality/DEV/SWAP-ERROR. (com.github.jcustenborder.kafka.connect.spooldir.AbstractCleanUpPolicy:90)
[2021-08-24 19:30:12,243] ERROR Exception encountered processing line 5692 of /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv. (com.github.jcustenborder.kafka.connect.spooldir.AbstractSourceTask:265)
at com.opencsv.CSVReader.readNext(CSVReader.java:353)
[2021-08-24 19:30:12,243] INFO Closing /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv (com.github.jcustenborder.kafka.connect.spooldir.InputFile:173)
[2021-08-24 19:30:12,245] ERROR Error during processing, moving /connect_integrations/848_quality/DEV/SWAP/SWAP-2021_06_0001_0.csv to /connect_integrations/848_quality/DEV/SWAP-ERROR. (com.github.jcustenborder.kafka.connect.spooldir.AbstractCleanUpPolicy:90)

Could you please help to identify where is the problem?

Confluent kafka 6.1.0 (Kafka version 2.7.1)
Used jcustenborder-kafka-connect-spooldir-2.0.62 version

Connector configuration:

{
"name" : "csv-file-quality-test-orders-v0",
"config" : {
"connector.class" : "com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector",
"csv.first.row.as.header" : "true",
"csv.separator.char" : "59",
"empty.poll.wait.ms" : "6000",
"error.path" : "/connect_integrations/848_quality/DEV/SWAP-ERROR",
"finished.path" : "/connect_integrations/848_quality/DEV/SWAP-SUCCESS",
"halt.on.error" : "false",
"input.file.pattern" : ".*.csv",
"input.path" : "/connect_integrations/848_quality/DEV/SWAP",
"key.converter" : "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable" : "false",
"name" : "csv-file-quality-test-orders-v0",
"output.json.formatter" : "com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson",
"topic" : "csv_test_topic",
"value.converter" : "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable" : "false",
"value.schema" : "{"name": "csv.Value", "type": "STRUCT", "isOptional": true, "fieldSchemas": {"test0": { "type": "STRING", "isOptional": true} , "test1" : { "type": "STRING", "isOptional": true} , "test2" : { "type": "STRING", "isOptional": true} , "test3" : { "type": "STRING", "isOptional": true} , "test4" : { "type": "STRING", "isOptional": true} , "test5" : { "type": "STRING", "isOptional": true} , "test6" : { "type": "STRING", "isOptional": true} , "test7" : { "type": "STRING", "isOptional": true} , "extended" : { "type": "STRING", "isOptional": true} , "extended1" : { "type": "STRING", "isOptional": true} , "extended2" : { "type": "STRING", "isOptional": true} , "extended3" : { "type": "STRING", "isOptional": true} , "extended4" : { "type": "STRING", "isOptional": true} , "extended5" : { "type": "STRING", "isOptional": true} , "extended6" : { "type": "STRING", "isOptional": true} , "extended7" : { "type": "STRING", "isOptional": true} , "extended8" : { "type": "STRING", "isOptional": true} , "extended9" : { "type": "STRING", "isOptional": true} , "extended10" : { "type": "STRING", "isOptional": true} , "extended11" : { "type": "STRING", "isOptional": true} , "extended12" : { "type": "STRING", "isOptional": true} , "extended13" : { "type": "STRING", "isOptional": true} , "extended14" : { "type": "STRING", "isOptional": true} , "extended15" : { "type": "STRING", "isOptional": true} , "extended16" : { "type": "STRING", "isOptional": true} , "extended17" : { "type": "STRING", "isOptional": true} , "extended18" : { "type": "STRING", "isOptional": true} , "extended19" : { "type": "STRING", "isOptional": true} , "extended20" : { "type": "STRING", "isOptional": true} , "extended21" : { "type": "STRING", "isOptional": true} , "extended22" : { "type": "STRING", "isOptional": true} , "extended23" : { "type": "STRING", "isOptional": true} , "extended24" : { "type": "STRING", "isOptional": true} , "extended25" : { "type": "STRING", "isOptional": true} , "extended26" : { "type": "STRING", "isOptional": true} , "extended27" : { "type": "STRING", "isOptional": true} , "extended28" : { "type": "STRING", "isOptional": true} , "extended29" : { "type": "STRING", "isOptional": true} , "extended30" : { "type": "STRING", "isOptional": true} , "extended31" : { "type": "STRING", "isOptional": true} } }"
}
}

Thank you.

[cande1gut]

@MaxSamsonov I had the same problem, what worked for me was to change the parser to RFC 4180 and change the file's name convention to camel case.