Schema-registry: Provide option to output json when using AvroConverter with kafka-connect

[holgerbrandl]

Currently when using a httpsink connector such as

curl -s -X POST -H  "Content-Type:application/json" \
    http://localhost:8083/connectors/ \
    -d '
{
  "name": "sensor_avro_sink",
  "config": {
    "topics.regex": "SOME_TOPIC",
    "tasks.max": "1",
    "connector.class": "io.confluent.connect.http.HttpSinkConnector",
    "http.api.url": "http://localhost:8085",
    "request.method": "post",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "consumer.override.auto.offset.reset": "latest",
    "confluent.topic.replication.factor": "1"
  }
}' 

to feed a REST API, the received body format is always _just_ Struct such as

Struct{TEMP=6,COUNTER=3}

However, it would be much better -- living in a REST world -- if a more standardized format such as json could be supported as well by AvroConverter. Currently, we need to duplicate topics (where we convert from avro to json) just for the sake of using connect.

[OneCricketeer]

Can you please clarify why you're not using the JSONConverter?

Or why you cannot write a SMT to do what you're saying?

[holgerbrandl]

I've tried to use JSON Converter with the config

{
  "name": "sensor_avro_sink",
  "config": {
    "topics.regex": "TEST_AVRO",
    "tasks.max": "1",
    "connector.class": "io.confluent.connect.http.HttpSinkConnector",
    "http.api.url": "http://localhost:8085",
    "request.method": "post",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "consumer.override.auto.offset.reset": "latest",
    "confluent.topic.replication.factor": "1"
  }
}

but this failed with the following error:

org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error:
at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:355)
at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:86)
at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$2(WorkerSinkTask.java:488)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)\n\t... 13 more\nCaused by: org.apache.kafka.common.errors.SerializationException: java.io.CharConversionException: Invalid UTF-32 character 0x1b010602 (above 0x0010ffff)
at char #1, byte #7)\nCaused by: java.io.CharConversionException: Invalid UTF-32 character 0x1b010602 (above 0x0010ffff)

When using the AvroConverter at least the avro conversion works, but the output format ist _just_ Struct (whatever that is technically).

About the SMT: I'm not aware of an existing SMT that would take an AVRO encoded topic and emit its value as a json to an HttpSinkConnector. One motivation on my end to file this ticket was to clarify if an HttpSinkConnector can be connected to an Avro ticket with config only without writing custom code.

[OneCricketeer]

is just Struct (whatever that is technically).

Its in internal Kafka Connect object type for serialization. Just because data is produced in Avro, it's only stored as binary in Kafka. The JSONConverter should be able to be used to deserialize those bytes.

The problem you're seeing is that the HTTP sink appears to follow the same logic as the FileSink; it only runs a toString on the Struct, and therefore you either must address the issue in that connector (because the problem has nothing to do with Avro or the Schema Registry), or you must write custom code (such as an SMT). Note: schemas.enable nor the schema registry url properties are necessary for the use case of getting JSON events

[OneCricketeer]

I'm not aware of an existing SMT that would take an AVRO encoded topic

Me neither, but again, Avro isn't the problem.

The data flow goes like so

topic -> converter -> Schema+Struct -> [SMT,...] -> Sink

[holgerbrandl]

Thanks for the nice explanation, in particular, the flow graph.

I've checked https://docs.confluent.io/current/connect/transforms/index.html and it seems that none of them deals with json, so I'd indeed need to write a custom transformation to implement the described feature. Imho this should be part of the platform since avro-topic -> rest-sink seems a rather standard use-case (at least to me).

[OneCricketeer]

FWIW, Avro also works over HTML, so the fact that you want JSON data is seemingly misguided

[holgerbrandl]

Why would be misguided to use a REST API with a json payload? That seems to be a much more common one than sending Avro encoded binary payloads via HTTP.

[rgannu]

I am also facing this problem. I am having the DB changes as CDC events (using Debezium kafka source connector) messages in AVRO format. I need the messages to be in AVRO to have concise message size. Now, I need for an usecase to send the AVRO messages to AMQP brokers via JMS.

JMS sink connector (camel-amqp-sink-connector) is unable to transfer and it gives the error that it is unable to convert type:

 Caused by: javax.jms.MessageFormatException: No type converter available to convert from type: org.apache.kafka.connect.data.Struct to the required type: java.io.Serializable with value Struct{...}

Definitely it is needed to convert the AVRO to JSON before sending to AMQP broker as JMS messages. Is there a SMT available for converting the AVRO to JSON ?

[OneCricketeer]

it is needed to convert the AVRO to JSON ... to AMQP

Why? Does JMS not accept byte[]?

[OneCricketeer]

Also, AvroDeserializer does output JSON. Have you never used kafka-avro-console-consumer?

The converter wraps the deserializer, so this issue, in context of the converter doesn't make sense to me.

The errors you've been posted are all downstream and out of scope of the converter's lifecycle

You can use JSONConverter in a sink connector, even if the producer to the topic was using Avro. This is because Connect does not know anything about the data format (because Kafka only stores bytes), it only knows its internal Schema and Struct types.

[OneCricketeer]

I've checked https://docs.confluent.io/current/connect/transforms/index.html and it seems that none of them deals with json, so I'd indeed need to write a custom transformation to implement the described feature

You could also use KSQL or Kafka Streams to create an intermediate, persistent JSON topic rather than convert in-memory.

[dinegri]

I also need this and I am thinking to implement it

but before that I really want to believe you when you said “You can use JSONConverter in a sink connector, even if the producer to the topic was using Avro”

If I use a AvroConveter, it contains an internal AvroDeserializer, the only way to unwrap a data generate by an AvroSerializer is using a AvroDeserializer

Will JSONConverter ‘unwrap’ a data serialized by an AvroSerializer?

Sure Kafka Connect has its only data struct but only after convert the data, and the converter has its internal Serializer/Deserializer

By the way, ksqldb does not have support
to TopicRecordNameStrategy yet

[OneCricketeer]

Example: Avro serializer produces data. Output: Filesink connector with no file path will print out to console. Use JsonConverter in that sink connector, and it'll be JSON. No errors because input was originally Avro. Try it

[sangvk]

We have exactly similar use case where a Change Data Capture tool reads the database changelogs and publishes messages to kafka topic, we want to filter these AVRO messages and post the filtered messages as json to an existing REST endpoint. I tried using transformations like below in the HttpSinkConnector, it didn't work, the transformation is not happening. (We have seen this kind of transformation working for the kafka-jdbc-connectors). Any other way we can try to filter and post json?

"fields.whitelist": "ORDER_ID, ORDER_NUM", -- looking for a config like this as I don't want to post the entire AVRO message, just 2 fields
"transforms":"FilterMessages, FlattenMessage, ReplaceFields",
"transforms.FilterMessages.type": "io.confluent.connect.transforms.Filter$Value",
"transforms.FilterMessages.filter.condition": "$.ordorderdata[?(@.IS_EMERGENCY == 1)]", -- want to filter messages with a filter condition like this
"transforms.FilterMessages.filter.type": "include",
"transforms.FilterMessages.missing.or.null.behavior": "exclude",
"transforms.FlattenMessage.type": "org.apache.kafka.connect.transforms.Flatten$Value", -- flattening AVRO structure
"transforms.ReplaceFields.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.ReplaceFields.renames": "ordorderdata.ORDER_ID:ORDER_ID, ordorderdata.ORDER_NUM:ORDER_NUM"

[dinegri]

Http sink connector has launched an option to output in json using AvroConverter

Please check the latest version, about transformation I cannot help you

[sangvk]

request.body.format=json seems working, thank you for the reply.