Connector failing while using @ in the schema

[shaktipanda]

Hi,

I am developing a sfdcplatformevent source connector where the configuration is

{
"name": "SFDC_scenariomatrix_billing_source",
"config": {
"connector.class": "io.confluent.salesforce.SalesforcePlatformEventSourceConnector",
"tasks.max": "1",
"kafka.topic":"tef.de.scenariomatrix-B2B-v1",
"kafka.topic.lowercase":"false",
"transforms": "ExtractField,fromJson",
"transforms.ExtractField.type":"org.apache.kafka.connect.transforms.ExtractField$Value",
"transforms.ExtractField.field":"TextArea__c",
"transforms.fromJson.type" : "com.github.jcustenborder.kafka.connect.json.FromJson$Value",
"transforms.fromJson.json.schema.location" : "Inline",
"transforms.fromJson.json.schema.inline" : "{"type":"object","properties":{"eventId":{"type":"string"},"eventTime":{"type":"string"},"eventType":{"type":"string"},"domain":{"type":"string"},"@type":{"type":"string"},"@basetype":{"type":"string"},"event":{"type":"object","properties":{"scenarioMatrix":{"type":"object","properties":{"id":{"type":"string"},"href":{"type":"string"},"@type":{"type":"string"}}}}}}}",
"value.converter":"io.confluent.connect.avro.AvroConverter",
"key.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.ssl.keystore.password":"mystorepassword",
"key.converter.schema.registry.ssl.keystore.password":"mystorepassword",
"key.converter.schema.registry.ssl.truststore.password":"mystorepassword",
"value.converter.schema.registry.url":"https://schemaregistry.ict.svc.cluster.local:8081",
"key.converter.schema.registry.ssl.keystore.location":"/mnt/sslcerts/keystore.p12",
"value.converter.schema.registry.ssl.truststore.password":"mystorepassword",
"key.converter.schema.registry.ssl.truststore.location":"/mnt/sslcerts/truststore.p12",
"value.converter.schema.registry.ssl.keystore.location":"/mnt/sslcerts/keystore.p12",
"value.converter.schema.registry.ssl.truststore.location":"/mnt/sslcerts/truststore.p12",
"key.converter.schema.registry.url":"https://schemaregistry.ict.svc.cluster.local:8081",
"salesforce.instance": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_URL}",
"salesforce.username": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_USERNAME}",
"salesforce.password": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_PASSWORD}",
"salesforce.password.token": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_PASSWORD_TOKEN}",
"http.proxy": "10.169.127.8:8080",
"salesforce.consumer.key": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_CONSUMER_KEY}",
"salesforce.consumer.secret": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_CONSUMER_SECRET}",
"salesforce.initial.start": "all",
"salesforce.platform.event.name": "TEF_ScenarioMatrix__e"
}
}

when schema in the above configuration is used with @ we get the following error

nCaused by: org.apache.avro.SchemaParseException: Illegal initial character: @basetype\n\tat org.apache.avro.Schema.validateName(Schema.java:1567)\n\tat org.apache.avro.Schema.access$400(Schema.java:92)\n\tat org.apache.avro.Schema$Field.(Schema.java:549)\n\tat org.apache.avro.Schema$Field.(Schema.java:588)\n\tat io.confluent.connect.avro.AvroData.addAvroRecordField(AvroData.java:1175)\n\tat io.confluent.connect.avro.AvroData.fromConnectSchema(AvroData.java:952)\n\tat io.confluent.connect.avro.AvroData.fromConnectSchema(AvroData.java:760)\n\tat io.confluent.connect.avro.AvroData.fromConnectSchema(AvroData.java:745)\n\tat io.confluent.connect.avro.AvroConverter.fromConnectData(AvroConverter.java:85)\n\tat org.apache.kafka.connect.storage.Converter.fromConnectData(Converter.java:63)\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$convertTransformedRecord$4(WorkerSourceTask.java:333)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:166)\n\tat org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:200)\n\t... 11 more\n"}]

But when we use the below configuratin where there is no @ in the schema

{
"name": "SFDC_scenariomatrix_billing_source",
"config": {
"connector.class": "io.confluent.salesforce.SalesforcePlatformEventSourceConnector",
"tasks.max": "1",
"kafka.topic":"tef.de.scenariomatrix-B2B-v1",
"kafka.topic.lowercase":"false",
"transforms": "ExtractField,fromJson",
"transforms.ExtractField.type":"org.apache.kafka.connect.transforms.ExtractField$Value",
"transforms.ExtractField.field":"TextArea__c",
"transforms.fromJson.type" : "com.github.jcustenborder.kafka.connect.json.FromJson$Value",
"transforms.fromJson.json.schema.location" : "Inline",
"transforms.fromJson.json.schema.inline" : "{"type":"object","properties":{"eventId":{"type":"string"},"eventTime":{"type":"string"},"eventType":{"type":"string"},"domain":{"type":"string"},"type":{"type":"string"},"baseType":{"type":"string"},"event":{"type":"object","properties":{"scenarioMatrix":{"type":"object","properties":{"id":{"type":"string"},"href":{"type":"string"},"type":{"type":"string"}}}}}}}",
"value.converter":"io.confluent.connect.avro.AvroConverter",
"key.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.ssl.keystore.password":"mystorepassword",
"key.converter.schema.registry.ssl.keystore.password":"mystorepassword",
"key.converter.schema.registry.ssl.truststore.password":"mystorepassword",
"value.converter.schema.registry.url":"https://schemaregistry.ict.svc.cluster.local:8081",
"key.converter.schema.registry.ssl.keystore.location":"/mnt/sslcerts/keystore.p12",
"value.converter.schema.registry.ssl.truststore.password":"mystorepassword",
"key.converter.schema.registry.ssl.truststore.location":"/mnt/sslcerts/truststore.p12",
"value.converter.schema.registry.ssl.keystore.location":"/mnt/sslcerts/keystore.p12",
"value.converter.schema.registry.ssl.truststore.location":"/mnt/sslcerts/truststore.p12",
"key.converter.schema.registry.url":"https://schemaregistry.ict.svc.cluster.local:8081",
"salesforce.instance": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_URL}",
"salesforce.username": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_USERNAME}",
"salesforce.password": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_PASSWORD}",
"salesforce.password.token": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_PASSWORD_TOKEN}",
"http.proxy": "10.169.127.8:8080",
"salesforce.consumer.key": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_CONSUMER_KEY}",
"salesforce.consumer.secret": "${file:/mnt/secrets/connector-creds/sfdc.properties:SALESFORCE_CONSUMER_SECRET}",
"salesforce.initial.start": "all",
"salesforce.platform.event.name": "TEF_ScenarioMatrix__e"
}
}

the connector runs fine without giving any error .

So can u please tell us that how can we resolve this issue of @ and if @ will be supported by kafka connector or not