Convert NettyChannelBufferStreamCache from NettyHttpSource not converted to string

[h4gen]

Hi everybody,

This Issue was originally posted by me in Zulip

I am currently trying to accept text or json messages via the CamelNettyhttpSourceConnector component.
I use the following configuration:

{
 "connector.class": "org.apache.camel.kafkaconnector.nettyhttp.CamelNettyhttpSourceConnector",
 "camel.source.endpoint.receiveBufferSize": "10",
 "camel.source.path.host": "0.0.0.0",
 "camel.source.path.protocol": "http",
 "tasks.max": "1",
 "topics": "test",
 "camel.source.path.port": "8081",
 "name": "http",
 "value.converter": "org.apache.kafka.connect.storage.StringConverter",
 "key.converter": "org.apache.kafka.connect.storage.StringConverter"
}

I send messages to the connector with:

curl -H "Content-Type: text/plain" -X POST -d '{"username":"xyz","password":"xyz"}' http://localhost:8081

or

curl -H "Content-Type: application/json" -X POST -d '{"username":"xyz","password":"xyz"}' http://localhost:8081

The messages end up in my topic "test" but there is just a bunch of org.apache.camel.component.netty.http.NettyChannelBufferStreamCache Objects or sometimes even empty objects.

The error occurs with version 0.7.0 and 0.7.1. I just downloaded the tar archives from the components page in the respective versions and extracted them into my kafka connect connector folder (Cloudera CDP 7.1)

Thank you very much.

[oscerd]

Thanks for reporting. We need to provide a converter or smt, with mechanism to convert from stream to string.

[valdar]

@h4gen can you try to add this to your configuration:

transforms: tostring
transforms.tostring.type: org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform
transforms.tostring.target.type: java.lang.String

[h4gen]

Hi @valdar!
Thanks for the suggestion. I added the snippet to the configuration, leading to the following configuration in total:

{
 "connector.class": "org.apache.camel.kafkaconnector.nettyhttp.CamelNettyhttpSourceConnector",
 "camel.source.endpoint.receiveBufferSize": "10",
 "camel.source.path.host": "0.0.0.0",
 "camel.source.path.protocol": "http",
 "tasks.max": "1",
 "topics": "test",
 "camel.source.path.port": "8081",
 "name": "http",
 "value.converter": "org.apache.kafka.connect.storage.StringConverter",
 "key.converter": "org.apache.kafka.connect.storage.StringConverter",
 "transforms": "tostring",
 "transforms.tostring.type": "org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform",
 "transforms.tostring.target.type": "java.lang.String"
}

Now I receive the following Error when i try to deploy the connector:

transforms.tostring.type: Invalid value class org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform for configuration transforms.tostring.type: Error getting config definition from Transformation: null

I assume this was not the expected behavior? :)
Thanks!

[valdar]

@h4gen yes my bad, wrong config. Try:

transforms: tostring
transforms.tostring.type: org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value
transforms.tostring.target.type: java.lang.String

note the added $Value in the type.

[h4gen]

@valdar thanks for the fast reply. Now the configuration was accepted and the connector deployed. However when I send messages to the connector i get the following:

org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104) at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50) at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:315) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:240) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.kafka.connect.errors.DataException: CamelTypeConverter was not able to convert value `null` to target type of `String` at org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform.convertValueWithCamelTypeConverter(CamelTypeConverterTransform.java:57) at org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform.apply(CamelTypeConverterTransform.java:47) at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162) ... 11 more 

Interestingly messages are forwarded to my topic but are just "empty". Does this make sense?

[valdar]

@h4gen this should be fixed now in master so will be available starting form version 0.9.0

[h4gen]

@valdar Thank you very much for taking care of this! Looking forward to try it out :)

[valdar]

@h4gen actually it should be in 0.7.3 too, that should be released sometime next week 🎉

[h4gen]

Hey @valdar: Was this included in 0.7.3 ? Because I just tried to use the 0.7.3 component and the error is still there :(