When using MSK Connect with Confluent S3 Sink connector and GSR converter for JSON schema following error is seen in logs causing the connector deployment to fail
[Worker-] [2022-02-01 07:16:17,218] INFO [confluent-s3-sink|task-2] [Consumer clientId=connector-consumer-confluent-s3-sink-2, groupId=connect-confluent-s3-sink] Notifying assignor about the new Assignment(partitions=[]) (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:276)
[Worker-] [2022-02-01 07:16:17,218] INFO [confluent-s3-sink|task-2] [Consumer clientId=connector-consumer-confluent-s3-sink-2, groupId=connect-confluent-s3-sink] Adding newly assigned partitions: (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:288)
[Worker-] [2022-02-01 07:16:17,218] INFO [confluent-s3-sink|task-3] [Consumer clientId=connector-consumer-confluent-s3-sink-3, groupId=connect-confluent-s3-sink] Notifying assignor about the new Assignment(partitions=[]) (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:276)
[Worker-] [2022-02-01 07:16:17,219] INFO [confluent-s3-sink|task-3] [Consumer clientId=connector-consumer-confluent-s3-sink-3, groupId=connect-confluent-s3-sink] Adding newly assigned partitions: (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:288)
[Worker-] [2022-02-01 07:16:17,219] INFO [confluent-s3-sink|task-1] [Consumer clientId=connector-consumer-confluent-s3-sink-1, groupId=connect-confluent-s3-sink] Successfully synced group in generation Generation{generationId=5, memberId='connector-consumer-confluent-s3-sink-1-a8e3fd8a-962c-468f-ac78-1293e23677e8', protocol='range'} (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:754)
[Worker-] [2022-02-01 07:16:17,219] INFO [confluent-s3-sink|task-1] [Consumer clientId=connector-consumer-confluent-s3-sink-1, groupId=connect-confluent-s3-sink] Notifying assignor about the new Assignment(partitions=[]) (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:276)
[Worker-] [2022-02-01 07:16:17,219] INFO [confluent-s3-sink|task-1] [Consumer clientId=connector-consumer-confluent-s3-sink-1, groupId=connect-confluent-s3-sink] Adding newly assigned partitions: (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:288)
[Worker-] [2022-02-01 07:16:17,219] INFO [confluent-s3-sink|task-0] [Consumer clientId=connector-consumer-confluent-s3-sink-0, groupId=connect-confluent-s3-sink] Notifying assignor about the new Assignment(partitions=[friends-0]) (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:276)
[Worker-] [2022-02-01 07:16:17,220] INFO [confluent-s3-sink|task-0] [Consumer clientId=connector-consumer-confluent-s3-sink-0, groupId=connect-confluent-s3-sink] Adding newly assigned partitions: friends-0 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:288)
[Worker-] [2022-02-01 07:16:17,317] INFO [confluent-s3-sink|task-0] [Consumer clientId=connector-consumer-confluent-s3-sink-0, groupId=connect-confluent-s3-sink] Found no committed offset for partition friends-0 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:1354)
[Worker-] [2022-02-01 07:16:17,423] INFO [confluent-s3-sink|task-0] [Consumer clientId=connector-consumer-confluent-s3-sink-0, groupId=connect-confluent-s3-sink] Resetting offset for partition friends-0 to position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[(id: 1 rack: apse2-az2)], epoch=0}}. (org.apache.kafka.clients.consumer.internals.SubscriptionState:396)
[Worker-] [2022-02-01 07:16:18,056] ERROR [confluent-s3-sink|task-0] WorkerSinkTask{id=confluent-s3-sink-0} Error converting message value in topic 'friends' partition 0 at offset 0 and timestamp 1643695005904: null (org.apache.kafka.connect.runtime.WorkerSinkTask:547)
[Worker-] java.lang.NullPointerException
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.typeconverters.StructTypeConverter.lambda$getOrderedFields$0(StructTypeConverter.java:339)
[Worker-] at java.base/java.util.TimSort.countRunAndMakeAscending(TimSort.java:355)
[Worker-] at java.base/java.util.TimSort.sort(TimSort.java:220)
[Worker-] at java.base/java.util.Arrays.sort(Arrays.java:1515)
[Worker-] at java.base/java.util.stream.SortedOps$SizedRefSortingSink.end(SortedOps.java:353)
[Worker-] at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:485)
[Worker-] at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
[Worker-] at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
[Worker-] at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
[Worker-] at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:578)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.typeconverters.StructTypeConverter.getOrderedFields(StructTypeConverter.java:345)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.typeconverters.StructTypeConverter.toConnectSchema(StructTypeConverter.java:315)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.JsonSchemaToConnectSchemaConverter.toConnectSchema(JsonSchemaToConnectSchemaConverter.java:86)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.JsonSchemaToConnectSchemaConverter.toConnectSchema(JsonSchemaToConnectSchemaConverter.java:63)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.JsonSchemaConverter.toConnectData(JsonSchemaConverter.java:202)
[Worker-] at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.convertValue(WorkerSinkTask.java:545)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:501)
[Worker-] at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)
[Worker-] at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)
[Worker-] at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:501)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:478)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:328)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
[Worker-] at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)
[Worker-] at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:238)
[Worker-] at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
[Worker-] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[Worker-] at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[Worker-] at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[Worker-] at java.base/java.lang.Thread.run(Thread.java:829)
[Worker-] [2022-02-01 07:16:18,059] ERROR [confluent-s3-sink|task-0] WorkerSinkTask{id=confluent-s3-sink-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:191)
[Worker-] org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
[Worker-] at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:206)
[Worker-] at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:501)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:478)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:328)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
[Worker-] at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)
[Worker-] at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:238)
[Worker-] at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
[Worker-] at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
[Worker-] at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[Worker-] at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[Worker-] at java.base/java.lang.Thread.run(Thread.java:829)
[Worker-] Caused by: java.lang.NullPointerException
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.typeconverters.StructTypeConverter.lambda$getOrderedFields$0(StructTypeConverter.java:339)
[Worker-] at java.base/java.util.TimSort.countRunAndMakeAscending(TimSort.java:355)
[Worker-] at java.base/java.util.TimSort.sort(TimSort.java:220)
[Worker-] at java.base/java.util.Arrays.sort(Arrays.java:1515)
[Worker-] at java.base/java.util.stream.SortedOps$SizedRefSortingSink.end(SortedOps.java:353)
[Worker-] at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:485)
[Worker-] at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
[Worker-] at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
[Worker-] at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
[Worker-] at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:578)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.typeconverters.StructTypeConverter.getOrderedFields(StructTypeConverter.java:345)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.typeconverters.StructTypeConverter.toConnectSchema(StructTypeConverter.java:315)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.JsonSchemaToConnectSchemaConverter.toConnectSchema(JsonSchemaToConnectSchemaConverter.java:86)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.JsonSchemaToConnectSchemaConverter.toConnectSchema(JsonSchemaToConnectSchemaConverter.java:63)
[Worker-] at com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.JsonSchemaConverter.toConnectData(JsonSchemaConverter.java:202)
[Worker-] at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.convertValue(WorkerSinkTask.java:545)
[Worker-] at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:501)
[Worker-] at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)
[Worker-] at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)
[Worker-] ... 13 more
[Worker-] [2022-02-01 07:16:18,061] INFO [confluent-s3-sink|task-0] [Consumer clientId=connector-consumer-confluent-s3-sink-0, groupId=connect-confluent-s3-sink] Revoke previously assigned partitions friends-0 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:307)
[Worker-] [2022-02-01 07:16:18,061] INFO [confluent-s3-sink|task-0] [Consumer clientId=connector-consumer-confluent-s3-sink-0, groupId=connect-confluent-s3-sink] Member connector-consumer-confluent-s3-sink-0-7693a472-d7ba-4458-8599-f37acb0cef3e sending LeaveGroup request to coordinator (id: 2147483646 rack: null) due to the consumer is being closed (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:1038)
connector.class=io.confluent.connect.s3.S3SinkConnector
value.converter.schemaAutoRegistrationEnabled=true
s3.region=ap-southeast-2
schema.compatibility=BACKWARD
flush.size=2
tasks.max=1
key.converter.region=ap-southeast-2
value.converter.dataFormat=JSON
value.converter.registry.name=testregistry
value.converter.endpoint=https://glue.ap-southeast-2.amazonaws.com
topics.regex=friends*
value.converter.avroRecordType=GENERIC_RECORD
value.converter.region=ap-southeast-2
format.class=io.confluent.connect.s3.format.json.JsonFormat
value.converter.schemaName=friends
name=confluent-s3-sink
value.converter.schemas.enable=true
value.converter=com.amazonaws.services.schemaregistry.kafkaconnect.jsonschema.JsonSchemaConverter
storage.class=io.confluent.connect.s3.storage.S3Storage
s3.bucket.name=msk-cluster-backup-deepens
key.converter=org.apache.kafka.connect.storage.StringConverter
Received message: key = message-0, value = JsonDataWithSchema(schema={"$id":"https://example.com/person.schema.json","$schema":"http://json-schema.org/draft-07/schema#","title":"Person","type":"object","properties":{"firstName":{"type":"string","description":"The person's first name."},"lastName":{"type":"string","description":"The person's last name."},"age":{"description":"Age in years which must be equal to or greater than zero.","type":"integer","minimum":0}}}, payload={"firstName":"John","lastName":"Doe","age":12})