camel-sftp-connector with k8s Strimzi kafka

[uiseong12]

I want to read json file through sftp connector.

source-connector.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
name: camel-sftp-source-kafka-connector
labels:
strimzi.io/cluster: my-connect-cluster
spec:
class: org.apache.camel.kafkaconnector.sftpsource.CamelSftpsourceSourceConnector
tasksMax: 1
config:
topics: 222
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: false
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: false
camel.kamelet.sftp-source.connectionHost: 192.168.1.170
camel.kamelet.sftp-source.connectionPort: 22
camel.source.endpoint.delete: true
camel.kamelet.sftp-source.directoryName: /test1
camel.kamelet.sftp-source.password: test
camel.kamelet.sftp-source.username: test

Error message

2023-01-18 15:04:31,151 INFO WorkerSourceTask{id=camel-sftp-source-kafka-connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask) [task-thread-camel-sftp-source-kafka-connector-0] 2023-01-18 15:04:31,151 ERROR WorkerSourceTask{id=camel-sftp-source-kafka-connector-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask) [task-thread-camel-sftp-source-kafka-connector-0] org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:206) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132) at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:298) at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:324) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:248) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:182) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:231) 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) Caused by: org.apache.kafka.connect.errors.DataException: Invalid type for bytes type: class org.apache.camel.converter.stream.ByteArrayInputStreamCache at org.apache.kafka.connect.json.JsonConverter.convertToJson(JsonConverter.java:653) at org.apache.kafka.connect.json.JsonConverter.convertToJsonWithEnvelope(JsonConverter.java:592) at org.apache.kafka.connect.json.JsonConverter.fromConnectData(JsonConverter.java:346) at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63) at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$2(WorkerSourceTask.java:298) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190) ... 11 more 2023-01-18 15:04:31,151 INFO Stopping CamelSourceTask connector task (org.apache.camel.kafkaconnector.CamelSourceTask) [task-thread-camel-sftp-source-kafka-connector-0] 2023-01-18 15:04:31,151 INFO Apache Camel 3.18.2 (camel-3) is shutting down (timeout:45s) (org.apache.camel.impl.engine.AbstractCamelContext) [task-thread-camel-sftp-source-kafka-connector-0]

what's means 'org.apache.camel.converter.stream.ByteArrayInputStreamCache'?

[scholzj]

It is not easy to read the error and the config without proper formatting. I think it does not understand the data format it receives. But I'm not sure how many Camel folks are around who have proper knowledge of what the connector expects. Questions like this are usually better asked to the connector authors since this differs connector by connector.