Redpanda connector not working.

[cgokulnath]

httpsinkconnector.log

I am trying to use kafka comptabile redpanda for http sink connector with the configuraiton as

{
"name": "http-sink-test",
"config": {
"connector.class": "io.aiven.kafka.connect.http.HttpSinkConnector",
"topics.regex": "abc",
"http.authorization.type": "oauth2",
"http.url": "https://xxx/data",
"http.headers.content.type": "application/json",
"oauth2.access.token.url": "https://xxx/token",
"oauth2.client.id": "yyyy",
"oauth2.client.secret": "zzz",
"oauth2.client.authorization.mode": "header",
"oauth2.response.token.property": "access_token",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter"
}
} and get the following error in status of the connector.

"tasks": [
{
"id": 0,
"state": "FAILED",
"worker_id": "10.42.3.119:8083",
"trace": "org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:230)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:156)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:522)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:499)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:335)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)\n\tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:833)\nCaused by: org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error: \n\tat org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:326)\n\tat org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:88)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$4(WorkerSinkTask.java:522)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:180)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:214)\n\t... 14 more\nCaused by: org.apache.kafka.common.errors.SerializationException: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'Message': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')\n at [Source: (byte[])"Message 100"; line: 1, column: 9]\n\tat org.apache.kafka.connect.json.JsonDeserializer.deserialize(JsonDeserializer.java:66)\n\tat org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:324)\n\t... 18 more\nCaused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'Message': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')\n at [Source: (byte[])"Message 100"; line: 1, column: 9]\n\tat com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:2391)\n\tat com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:745)\n\tat com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3635)\n\tat com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2734)\n\tat com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:902)\n\tat com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:794)\n\tat com.fasterxml.jackson.databind.ObjectMapper._readTreeAndClose(ObjectMapper.java:4703)\n\tat com.fasterxml.jackson.databind.ObjectMapper.readTree(ObjectMapper.java:3090)\n\tat org.apache.kafka.connect.json.JsonDeserializer.deserialize(JsonDeserializer.java:64)\n\t... 19 more\n"
}

it is to be noted that curl command inside the http-sink-pod works correctly.

curl --location 'https://mvp-preview2-idp-ra.centralus.cloudapp.azure.com:443/idp/timeseries/v1/b8a18df1-8840-4156-9897-0801ed52c66d/sources/69060a7b-f7a7-4289-a64d-6658d913b86b/data'
--header 'Content-Type: application/json'
--header 'Authorization: Bearer tokenvalue'
--data '{
"value": 4,
"quality": 192,
"status": "OK",
"timeStamp": "2023-08-10T15:52:52.116Z",
"annotations": {
"productColor": "red",
"productCode": "765665.1"
}
}

can you please assist here on what went wrong here. do i need to change anything on configuration

[ivanyu]

Hi @cgokulnath
It seems one of the values in the topic cannot be serialized as JSON.

[cgokulnath]

Thanks @ivanyu . i corrected the timestamp that caused the json parsing error. now, i am getting this error,
httpsinkconnector.log
"trace": "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)\n\tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:833)\nCaused by: org.apache.kafka.connect.errors.ConnectException: Sending failed and no retries remain, stopping\n\tat io.aiven.kafka.connect.http.HttpSinkTask.put(HttpSinkTask.java:112)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)\n\t... 11 more\n"
}

[cgokulnath]

httpsinkconnector.log
{
"http-sink-test": {
"info": {
"name": "http-sink-test",
"config": {
"connector.class": "io.aiven.kafka.connect.http.HttpSinkConnector",
"http.authorization.type": "oauth2",
"oauth2.client.id": "abc",
"oauth2.client.secret": "ade",
"http.headers.content.type": "application/json",
"http.url": "xxx/data",
"oauth2.response.token.property": "access_token",
"topics.regex": "simutopic*",
"oauth2.access.token.url": "yyy",
"oauth2.client.authorization.mode": "header",
"name": "http-sink-test",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"key.converter": "org.apache.kafka.connect.storage.StringConverter"
},
"tasks": [
{
"connector": "http-sink-test",
"task": 0
}
],
"type": "sink"
},
"status": {
"name": "http-sink-test",
"connector": {
"state": "RUNNING",
"worker_id": "10.42.3.126:8083"
},
"tasks": [
{
"id": 0,
"state": "FAILED",
"worker_id": "10.42.3.126:8083",
"trace": "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)\n\tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:833)\nCaused by: org.apache.kafka.connect.errors.ConnectException: Sending failed and no retries remain, stopping\n\tat io.aiven.kafka.connect.http.HttpSinkTask.put(HttpSinkTask.java:112)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)\n\t... 11 more\n"
}
],
"type": "sink"
}
}
}

[ivanyu]

Sending failed and no retries remain, stopping indicates that there's some issue with sending the HTTP requests. You need to check the earlier logs to get find the error.

[cgokulnath]

thanks ivan. the error seems to be with authentication used. the http end point i am using does not support 'client_credentails' in its post API. so, i tried to use hard coded token in the connector config as below.
{
"name": "http-sink-connector",
"config": {
"connector.class": "io.aiven.kafka.connect.http.HttpSinkConnector",
"tasks.max": "1",
"topics": "simutopic",
"http.url": "xxx",
"http.authorization.type": "static", // or "none" if you manage authorization manually
"http.headers.content.type": "application/json",
"retry.backoff.ms": "500",
"retry.max.time.ms": "3000",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter.schemas.enable": "false",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"key.converter.schemas.enable": "false",
"http.headers.authorization": "Bearer token"
}
}

i am getting error
the org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)\n\tat org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base/java.lang.Thread.run(Thread.java:833)\nCaused by: org.apache.kafka.connect.errors.ConnectException: Sending failed and no retries remain, stopping\n\tat io.aiven.kafka.connect.http.HttpSinkTask.put(HttpSinkTask.java:112)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)\n\t... 11 more\n

[shauryagoel]

Hi @cgokulnath, I am trying the same thing in redpanda and getting a warning like this-

{"ts":"2024-05-01 08:05:29,727","level":"WARN","thread":"main","class":"org.apache.kafka.connect.runtime.isolation.Plugins:122","msg":"One or more plugins are 
missing ServiceLoader manifests may not be usable with plugin.discovery=service_load: [                                                                        
classpath       com.redpanda.connectors.transforms.RedpandaCast$Key     transformation  undefined                                                              
classpath       com.redpanda.connectors.transforms.RedpandaCast$Value   transformation  undefined                                                              
classpath       com.redpanda.connectors.transforms.ReplaceTimestamp$Key transformation  undefined                                                              
classpath       com.redpanda.connectors.transforms.ReplaceTimestamp$Value       transformation  undefined                                                      
classpath       com.redpanda.connectors.transforms.SchemaRegistryReplicator     transformation  undefined                                                      
]                                                                                                                                                              
Read the documentation at https://kafka.apache.org/documentation.html#connect_plugindiscovery for instructions on migrating your plugins to take advantage of t
he performance improvements of service_load mode. To silence this warning, set plugin.discovery=only_scan in the worker config."}

No connector is showing in the redpanda console. How did you resolve this?