# SchemaAndValue to kafka connect struct not working for avro converter **URL:** \ **Category:** Kafka Connect\ **Created:** [14 November 2021 03:28 UTC](https://forum.confluent.io/t/schemaandvalue-to-kafka-connect-struct-not-working-for-avro-converter/3377 "2021-11-14T03:28:59Z")\ **Posts on this page:** 2\ **Page:** 1
**Author:** ![sam](https://avatars.discourse-cdn.com/v4/letter/s/5f9b8f/32.png) [@sam](https://forum.confluent.io/u/sam)\ **Post date:** [14 November 2021 03:28 UTC](https://forum.confluent.io/t/schemaandvalue-to-kafka-connect-struct-not-working-for-avro-converter/3377/1 "2021-11-14T03:28:59Z")
HI there: We are now using JsonConverter to convert the data from String to ShemaAndValue in SMT. It is a nested Schema and we can convert it to SchemaAndValue successfully. But when we use the schemaandvalue to build the schema and struct for SMT while using the value converter as avro converter. The Schema and value does not work. But when we use the schemabuilder to build the shcema and struct. It works fine. We compare both, the class, the content are same. But it just raise errors. Can anyone give us a hand. Thanks a Here is our example: ```auto val input = """{ "schema": { "type": "struct", "optional": false, "version": 1, "fields": [ { "field": "id", "type": "string", "optional": true }, { "field": "batters","type": "array", "items": { "type" : "struct", "fields": [{ "field": "bid", "type": "string" }, { "field": "type", "type": "string" }] }} ] }, "payload": { "id": "1", "batters": [{ "bid": "1001", "type":"parise" }] } }""".getBytes() converter.configure(Collections.emptyMap(), false) val schemaAndValue = converter.toConnectData("foo", input) ``` The method newRecord in SMT ```auto newRecord(record, schemaAndValue.schema,schemaAndValue .value ) ``` The errors: ```auto org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler connect | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178) connect | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:297) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:323) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:247) connect | at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184) connect | at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234) connect | at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) connect | at java.util.concurrent.FutureTask.run(FutureTask.java:266) connect | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) connect | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) connect | at java.lang.Thread.run(Thread.java:748) connect | Caused by: org.apache.kafka.connect.errors.DataException: Failed to serialize Avro data from topic array_nested_12 : connect | at io.confluent.connect.avro.AvroConverter.fromConnectData(AvroConverter.java:91) connect | at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$2(WorkerSourceTask.java:297) connect | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128) connect | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162) connect | ... 11 more connect | Caused by: org.apache.kafka.common.errors.SerializationException: Error serializing Avro message connect | Caused by: org.apache.avro.SchemaParseException: Can't redefine: io.confluent.connect.avro.ConnectDefault connect | at org.apache.avro.Schema$Names.put(Schema.java:1511) connect | at org.apache.avro.Schema$NamedSchema.writeNameRef(Schema.java:782) connect | at org.apache.avro.Schema$RecordSchema.toJson(Schema.java:943) connect | at org.apache.avro.Schema$ArraySchema.toJson(Schema.java:1102) connect | at org.apache.avro.Schema$RecordSchema.fieldsToJson(Schema.java:971) connect | at org.apache.avro.Schema$RecordSchema.toJson(Schema.java:955) connect | at org.apache.avro.Schemas.toString(Schemas.java:46) connect | at org.apache.avro.Schemas.toString(Schemas.java:30) connect | at io.confluent.kafka.schemaregistry.avro.AvroSchema.canonicalString(AvroSchema.java:140) connect | at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.registerAndGetId(CachedSchemaRegistryClient.java:206) connect | at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:268) connect | at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.register(CachedSchemaRegistryClient.java:244) connect | at io.confluent.kafka.serializers.AbstractKafkaAvroSerializer.serializeImpl(AbstractKafkaAvroSerializer.java:74) connect | at io.confluent.connect.avro.AvroConverter$Serializer.serialize(AvroConverter.java:143) connect | at io.confluent.connect.avro.AvroConverter.fromConnectData(AvroConverter.java:84) connect | at org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$2(WorkerSourceTask.java:297) connect | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128) connect | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162) connect | at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.convertTransformedRecord(WorkerSourceTask.java:297) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.sendRecords(WorkerSourceTask.java:323) connect | at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:247) connect | at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184) connect | at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234) connect | at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) connect | at java.util.concurrent.FutureTask.run(FutureTask.java:266) connect | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) connect | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) connect | at java.lang.Thread.run(Thread.java:748) ``` we try to compare the different ways to create the struct and schema(shceamabuilder and SchemaAndValue) ```auto val arrayvalueSchema = SchemaBuilder.struct().name("com.example.User") .field("bid", Schema.STRING_SCHEMA) .field("type", Schema.STRING_SCHEMA).build() val arrayschema = SchemaBuilder.array( arrayvalueSchema ).build() val builder = SchemaBuilder.struct.name("json_schema") .field("id",Schema.STRING_SCHEMA) .field("batters",arrayschema) val oldschema = builder.build() val arrayelementstruct1 = new Struct(arrayvalueSchema) .put("bid", "001" ) .put("type", "test1") val arrayelementstruct2 = new Struct(arrayvalueSchema) .put("bid", "002" ) .put("type", "test2") val array = util.Arrays.asList(arrayelementstruct1,arrayelementstruct2) val oldstruct = new Struct(oldschema) .put("id","01") .put("batters",array) // ---------------------------------------------------------------------- val converter = new JsonConverter() val input = """{ "schema": { "type": "struct", "optional": false, "version": 1, "fields": [ { "field": "id", "type": "string", "optional": true }, { "field": "batters","type": "array", "items": { "type" : "struct", "fields": [{ "field": "bid", "type": "string" }, { "field": "type", "type": "string" }] }} ] }, "payload": { "id": "1", "batters": [{ "bid": "1001", "type":"parise" }] } }""".getBytes() converter.configure(Collections.emptyMap(), false) val schemaAndValue = converter.toConnectData("foo", input) val jsShcmea = converter.asJsonSchema(schemaAndValue.schema()) // val struct = requireStruct(schemaAndValue.value(),"testing") val schema = schemaAndValue.schema() ``` The results are same ![ShceamBuildVSchemaAndValue](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/2X/d/d5b6b8385fde971c44a6f012324fd604a077999b.png) Many thanks again ---
**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\ **Post date:** [14 December 2021 03:29 UTC](https://forum.confluent.io/t/schemaandvalue-to-kafka-connect-struct-not-working-for-avro-converter/3377/2 "2021-12-14T03:29:32Z")
This topic was automatically closed 30 days after the last reply. New replies are no longer allowed.