Nested json with JSONSchema value converter DataException: Invalid null value for required STRUCT field #3101
Comments
|
Shouldn't you be sending |
|
You are right. The the error message I gave is coming, because I am missing the defined object. My real problem - and the error I got stuck on is when I am sending the message you also suggested is [2024-04-26 06:09:58,625] INFO Kafka startTimeMs: 1714111798620 (org.apache.kafka.common.utils.AppInfoParser)
[2024-04-26 06:09:58,938] INFO [Producer clientId=console-producer] Cluster ID: esaSwPz2R1GpJN7R-ALcIA (org.apache.kafka.clients.Metadata)
{"level0": {"level1": {"level2": 2}}} |
|
@rayokota Thanks for chipping in earlier. I solved the {
"type": "object",
"title": "titleA",
"properties": {
"level0": {
"type": "object",
"title": "titleB",
"properties": {
"level1": {
"type": "object",
"properties": {
"level2": {
"type": "integer"
}
}
}
}
}
},
"additionalProperties": false
}As soon as I remove any of the Can you point me to the direction, why this is happening or where can I learn more about mandatory JSON fields for connectors? I can also close the issue, as I found "a solution"TM, but ideally I would also know "why". |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
I am trying to send nested JSON messages and sink them using Kafka connect S3sink to parque.
Whenever I nest the object too much the connector gives me the error message below.
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104) at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:495) at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:472) at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:328) at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232) at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:204) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.kafka.connect.errors.DataException: Invalid null value for required STRUCT field at io.confluent.connect.json.JsonSchemaData.toConnectData(JsonSchemaData.java:511) at io.confluent.connect.json.JsonSchemaData.lambda$static$11(JsonSchemaData.java:192) at io.confluent.connect.json.JsonSchemaData.toConnectData(JsonSchemaData.java:561) at io.confluent.connect.json.JsonSchemaData.lambda$static$11(JsonSchemaData.java:192) at io.confluent.connect.json.JsonSchemaData.toConnectData(JsonSchemaData.java:561) at io.confluent.connect.json.JsonSchemaConverter.toConnectData(JsonSchemaConverter.java:108) at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87) at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:495) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162) ... 13 moreTo trigger the error I start a producer:
I send this message:
{"level0": {"k": {"i": 2}}}My connector config:
{ "store.url": "http://localstack:4566", "aws.access.key.id": "", "aws.secret.access.key": "", "connector.class": "io.confluent.connect.s3.S3SinkConnector", "flush.size": 1, "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat", "locale": "en-US", "name": "s3-sink", "parquet.codec": "snappy", "partition.duration.ms": 3600000, "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner", "rotate.interval.ms": 600000, "s3.bucket.name": "my-bucket", "s3.region": "eu-central-1", "storage.class": "io.confluent.connect.s3.storage.S3Storage", "tasks.max": 1, "timestamp.extractor": "Record", "timestamp.field": "time", "timezone": "UTC", "topics": "foo", "topics.dir": "topics", "value.converter": "io.confluent.connect.json.JsonSchemaConverter", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "value.converter.schema.registry.url": "http://schema-registry:8081" }I can do the nesting with AVRO without any problems.
The text was updated successfully, but these errors were encountered: