Http Sink Connector - encounter org.apache.kafka.connect.errors.ConnectException: Unknown value type: `java.util.HashMap`

[KevinMeng4662]

hi Team,

There is one exception during applying the io.lenses.streamreactor.connect.http.sink.HttpSinkConnector.
Could you help to check that?

Below is connector configuration:

`name=lenseshttp
connector.class=io.lenses.streamreactor.connect.http.sink.HttpSinkConnector
tasks.max=1
topics=topic.name
connect.http.authentication.type=none
connect.http.method=POST
connect.http.endpoint=https://XXXXXXX
connect.http.request.headers="Content-Type:application/cloudevents+json"
connect.http.request.content="{{value}}"
connect.http.batch.count=1

use the kafka converters that better suit your needs, these are just defaults:

key.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false`

Exception:

2024-11-14 15:31:19,543 ERROR [task-thread-lenseshttp-0] [io.lenses.streamreactor.connect.http.sink.tpl.substitutions.SubstitutionError$] [apply:27] SubstitutionError Raised: unable to extract field None for template, org.apache.kafka.connect.errors.ConnectException: Unknown value type: java.util.HashMap, string representation: '{data={gender=male, tenantId=tenantid, age=25}, specversion=1.0, source=da915f_87b9_414a_9f70_5f794a1310de, id=4d8f4a0f-cba5-45e4-9f9d-8168bfda77, time=2024-11-14T06:58:47.472733Z, type=event.type.v1, publishtime=2024-11-14T06:58:48.235556Z}', path: Empty at io.lenses.streamreactor.connect.cloud.common.sink.extractors.KafkaConnectExtractor$.extract(KafkaConnectExtractor.scala:69) at io.lenses.streamreactor.connect.cloud.common.sink.extractors.KafkaConnectExtractor$.extractFromValue(KafkaConnectExtractor.scala:38) at io.lenses.streamreactor.connect.http.sink.tpl.substitutions.Value$.get(Value.scala:24) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.TemplateRenderer.$anonfun$getTagValueFromData$6(TemplateRenderer.scala:85) at scala.util.Either.flatMap(Either.scala:360) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.TemplateRenderer.$anonfun$getTagValueFromData$2(TemplateRenderer.scala:81) at scala.util.Either.flatMap(Either.scala:360) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.TemplateRenderer.getTagValueFromData(TemplateRenderer.scala:72) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.TemplateRenderer.$anonfun$render$2(TemplateRenderer.scala:59) at scala.util.matching.Regex.$anonfun$replaceAllIn$1(Regex.scala:512) at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:619) at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:617) at scala.collection.AbstractIterator.foreach(Iterator.scala:1303) at scala.util.matching.Regex.replaceAllIn(Regex.scala:512) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.TemplateRenderer.$anonfun$render$1(TemplateRenderer.scala:56) at cats.syntax.EitherSyntax$CatchOnlyPartiallyApplied$.apply$extension(either.scala:52) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.TemplateRenderer.render(TemplateRenderer.scala:52) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.RecordRenderer$.renderRecord(RecordRenderer.scala:52) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.RecordRenderer$.$anonfun$renderRecords$1(RecordRenderer.scala:40) at scala.collection.immutable.List.map(List.scala:247) at scala.collection.immutable.List.map(List.scala:79) at cats.data.NonEmptySeq$.map$extension(NonEmptySeq.scala:191) at io.lenses.streamreactor.connect.http.sink.tpl.renderer.RecordRenderer$.renderRecords(RecordRenderer.scala:40) at io.lenses.streamreactor.connect.http.sink.tpl.SimpleTemplate.renderRecords(TemplateTypes.scala:66) at io.lenses.streamreactor.connect.http.sink.HttpSinkTask.$anonfun$processRecords$5(HttpSinkTask.scala:122) at scala.collection.immutable.RedBlackTree$.foreach(RedBlackTree.scala:291) at scala.collection.immutable.TreeMap.foreach(TreeMap.scala:195) at io.lenses.streamreactor.connect.http.sink.HttpSinkTask.processRecords(HttpSinkTask.scala:119) at io.lenses.streamreactor.connect.http.sink.HttpSinkTask.put(HttpSinkTask.scala:98) at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:601) at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:350) at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:250) at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:219) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:204) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:259) at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:236) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) at java.base/java.lang.Thread.run(Thread.java:833)

Thanks

[davidsloan]

The value is being converted to a HashMap on the way into the sink. When you use

connect.http.request.content="{{value}}"

you are asking for a String representation of a HashMap.

This can't be done in any sensible way, so the error occurs.

What needs to be done is either providing a key: {{value.mykey}} or alternatively to use a StringConverter on the Kafka record's value so that it is not converted into a Map - it will remain a json string.

[davidsloan]

I've been investigating this issue and I can see that a PR went in in January 2025 that should have resolved this:
54d1db5

Please could you try it with the latest version, and report back how you get on?