I am trying to learn how Kafka connector works and trying to use SftpCsvSourceConnector.
If the csv file has wrong data which fails in schema validation then csv file is moving to error folder as expected but connector became slow and not able to process next files.
I have added halt.on.error as false but still why it’s not able to proceed further.
This is the connector properties->
name=SftpSourceConnector
tasks.max=1
behavior.on.error=IGNORE
connector.class=io.confluent.connect.sftp.SftpCsvSourceConnector
cleanup.policy=MOVE
input.path=ftp
error.path=ftp/error
finished.path=ftp/archived
input.file.pattern=.*.csv
sftp.username=ftpuser
sftp.password=abcd1234
sftp.host=ftp-server
sftp.port=22
kafka.topic=kafka-sftp-demo
halt.on.error=false
csv.first.row.as.header=true
schema.generation.enabled=false
key.schema={\n "name" : "com.example.users.UserKey",\n "type" : "STRUCT",\n "isOptional" : false,\n "fieldSchemas" : {\n "id" : {\n "type" : "INT64",\n "isOptional" : false\n }\n }\n}
value.schema={\n "name" : "com.example.users.User",\n "type" : "STRUCT",\n "isOptional" : false,\n "fieldSchemas" : {\n "id" : {\n "type" : "INT64",\n "isOptional" : false\n },\n "first_name" : {\n "type" : "STRING",\n "isOptional" : true\n },\n "last_name" : {\n "type" : "STRING",\n "isOptional" : true\n },\n "email" : {\n "type" : "STRING",\n "isOptional" : true\n },\n "gender" : {\n "type" : "STRING",\n "isOptional" : true\n },\n "ip_address" : {\n "type" : "STRING",\n "isOptional" : true\n },\n "last_login" : {\n "type" : "STRING",\n "isOptional" : true\n },\n "account_balance" : {\n "name" : "org.apache.kafka.connect.data.Decimal",\n "type" : "BYTES",\n "version" : 1,\n "parameters" : {\n "scale" : "2"\n },\n "isOptional" : true\n },\n "country" : {\n "type" : "STRING",\n "isOptional" : true\n },\n "favorite_color" : {\n "type" : "STRING",\n "isOptional" : true\n }\n }\n}
This is the error log →
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
kafka-connect-configuration | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:266)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:293)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:228)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
kafka-connect-configuration | at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
kafka-connect-configuration | at java.util.concurrent.FutureTask.run(FutureTask.java:266)
kafka-connect-configuration | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
kafka-connect-configuration | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
kafka-connect-configuration | at java.lang.Thread.run(Thread.java:748)
kafka-connect-configuration | Caused by: org.apache.kafka.connect.errors.DataException: Conversion error: null value for field that is required and has no default value
kafka-connect-configuration | at org.apache.kafka.connect.json.JsonConverter.convertToJson(JsonConverter.java:586)
kafka-connect-configuration | at org.apache.kafka.connect.json.JsonConverter.convertToJson(JsonConverter.java:680)
kafka-connect-configuration | at org.apache.kafka.connect.json.JsonConverter.convertToJsonWithEnvelope(JsonConverter.java:567)
kafka-connect-configuration | at org.apache.kafka.connect.json.JsonConverter.fromConnectData(JsonConverter.java:320)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$1(WorkerSourceTask.java:266)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
kafka-connect-configuration | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)