How to produce messages with avro serde?..

[HyundeokPark]

Hi, I'm trying to produce messages with avro serde, but there is no option for avro like this,

Currently I'm using docket images with tag 'master' and in Topics - Messages section, I see Key/Value Serde have SchemaRegistry option

when i produce messages with String serde, my connectors are into failed task with
error below...

org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:223) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:149) at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:516) at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:493) at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:332) at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:234) at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:203) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:244) 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) Caused by: org.apache.kafka.connect.errors.DataException: Failed to deserialize data for topic labeled to Avro: at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:124) at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:88) at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$4(WorkerSinkTask.java:516) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:173) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:207) ... 13 more Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte! at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.getByteBuffer(AbstractKafkaSchemaSerDe.java:250) at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.<init>(AbstractKafkaAvroDeserializer.java:334) at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion(AbstractKafkaAvroDeserializer.java:202) at io.confluent.connect.avro.AvroConverter$Deserializer.deserialize(AvroConverter.java:172) at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:107) ... 17 more

is there somethin I missed?...
Thank you for yout answers in advance

P.S
I use strimzi for provisioning kafka, and kafka connect, connector plugins, also schema registry is provisioned by bitnami

[Amaralus]

hi, i have a same issue

[HyundeokPark]

I have figured it out my self, I didn't know that there are some rules

when you named schema subject name,

it should be form of topic's name-value or topic's name-key

if i have topic name 'label', then i should named 'label-value' for value serde

and same for the key

and this is the results I got

[iliax]

@HyundeokPark yes, you are right
kafka-ui only supports TopicNameStrategy for serialization now (see doc).
We are planning to add other policies in the future

[javapapo]

Hello not sure if this is related but I am currently running

v0.6.1

And I am still not getting Avro msg properly preview-ed

[iliax]

Hello @javapapo ,
Are you using SchemaRegistry for avro schemas storage? If so - you need to add SR-related properties to your configuration. (see basic config for sample.

When configured properly - you will be able to select "SchemaRegistry" serde list.

"Avro (Embedded)" serde is another type of serialization - when schema is embedded in message payload.

[javapapo]

Many thanks for the prompt reply - apologies I am slightly lost.

Do you mean this one?

schemaNameTemplate: how keys are saved to Schema Registry

I do have weird the SR to the basic config .

KAFKA_CLUSTERS_0_SCHEMA_REGISTRY: "https://xxxxxxxx-internal-eks.eu-xxxxxx"

And the integration is working..as far as I understand

[iliax]

@javapapo ok,

1. do you have "Schema Registry" line in left menu?
2. do you have "SchemaRegistry" serde option Value Serde dropdown (on your topic page)?

[javapapo]

1. Yes

2. No I only get this

And this is how it looks like from the Schema Registry side

Thanks for taking the time :)

[iliax]

@javapapo ok, I got it
can you please set KAFKA_CLUSTERS_0_SCHEMAREGISTRY property instead of KAFKA_CLUSTERS_0_SCHEMA_REGISTRY ? It should fix the issue

[javapapo]

Ooh! let me try!

[javapapo]

Legend! @iliax many many thanks for your time! Highly Appreciated!