Kafka-connect-jdbc: Support for schema prefixes.

[jcustenborder]

The schemas that are currently generated are based on the table name. When using the AvroConverter from the Confluent Schema Registry, the resulting avro schema does not have a namespace. This causes the avro-maven-plugin to attempt to place the generated source in default java package. The following exception is raised.

 error: `package' expected but `public' found.
[INFO] public class eng_match extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
[INFO] ^
[ERROR] /Users/raosa/gitprojects/examples/kafka-streams/target/generated-sources/eng_match.java:3619: error: identifier expected but eof found.
[INFO] }
[INFO]  ^
[ERROR] two errors found

[andrewrjones]

I'm having this issue to. Did you find a workaround?

[piotr-kalanski]

I'm also having this issue. It would be great to give possibility to provide target namespace...

[sarathkumar85]

Any workaround to fix the issue? is it the right approach to add the namespace to the generated schema and re-register?

[mageshn]

@sarathkumar85 would be great if you could some more details on how you are using the Schema generated by Connect. I would assume it's a downstream application(possibly streams) where you are trying to generate the SpecificRecord class based on the schema registered by connect?

[mageshn]

You could use a SMT to add namespace to the connect schema. Below is a sample SMT that helps add the namespace https://gist.github.com/vikas-tikoo-zefr/152100accce437adb258e5c6af031866

@vikas-tikoo-zefr

[terry19850829]

I have this issue. I have a table named "merchant". The schema kafka-jbdc-connect generateed is：
{"type":"record","name":"merchant","fields":[{"name":"merchant_id","type":"string"},{"name":"name","type":["null","string"],"default":null},{"name":"sn","type":["null","string"],"default":null},{"name":"status","type":["null","int"],"default":null},{"name":"owner_name","type":["null","string"],"default":null},{"name":"owner_cellphone","type":["null","string"],"default":null},{"name":"contact_name","type":["null","string"],"default":null},{"name":"contact_cellphone","type":["null","string"],"default":null},{"name":"contact_phone","type":["null","string"],"default":null},{"name":"reg_source","type":["null","string"],"default":null},{"name":"legal_person_register_no","type":["null","string"],"default":null},{"name":"business","type":["null","string"],"default":null},{"name":"legal_person_name","type":["null","string"],"default":null},{"name":"withdraw_mode","type":["null","int"],"default":null},{"name":"verify_status","type":["null","int"],"default":null},{"name":"organization_id","type":["null","string"],"default":null},{"name":"province","type":["null","string"],"default":null},{"name":"city","type":["null","string"],"default":null},{"name":"district","type":["null","string"],"default":null},{"name":"district_code","type":["null","string"],"default":null},{"name":"street_address","type":["null","string"],"default":null},{"name":"industry_id","type":["null","string"],"default":null},{"name":"longitude","type":["null","double"],"default":null},{"name":"latitude","type":["null","double"],"default":null},{"name":"user_id","type":["null","string"],"default":null},{"name":"ctime","type":["null","long"],"default":null},{"name":"mtime","type":["null","long"],"default":null},{"name":"version","type":["null","long"],"default":null},{"name":"industry_code","type":["null","string"],"default":null},{"name":"level1","type":["null","string"],"default":null},{"name":"level2","type":["null","string"],"default":null},{"name":"contact_email","type":["null","string"],"default":null},{"name":"crm_mtime","type":["null","long"],"default":null}],"connect.name":"merchant"}
I have a kstreams application downstream to join 2 streams. When I use the avro maven plugin generate the java class. The java class name is
public class merchant extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord
java compiler can't resolve it.

[jcustenborder]

@terry19850829 The comment above has the solution you can implement.

[terry19850829]

I have seen the solution. Thanks.

[kalbmj]

@jcustenborder @mageshn What is the process you used to get the SMT from @vikas-tikoo-zefr into your kafkaconnect definition? I am running into the same issue, and am unsure if this is still the best SMT to use as @terry19850829 and if so how the jar is created and added to the kafkaconnect classpath to solve the namespace issue. Thanks in advance.

[vikas-tikoo-zefr]

@kalbmj the SMT should still work. To use, add the following configs to your connector definition -

"transforms":"AddNamespace",
"transforms.AddNamespace.type":"<package_name>.Namespacefy",
"transforms.AddNamespace.timestamp.record.namespace":"<namespace_value>"

As far as deployment is concerned, you'll need to make a jar off it, and copy it onto your plugins path directory.

[kalbmj]

Thanks for the response @vikas-tikoo-zefr . I am new to custom SMTs, so sorry for all of the questions: 1. what is needed alongside the java file in order to create the jar? or should this be brought into an unbuilt kafka-connect-jdbc jar and built there? 2. is there an example you know of anywhere for building the SMT jars (or pom examples)? Thanks.

[vikas-tikoo-zefr]

This can be a jar on its own. For dependenciesorg.apache.kafka:kafka-clients, org.apache.kafka:connect-api and org.apache.kafka:connect-transforms; should mostly suffice.

[kalbmj]

Thanks @vikas-tikoo-zefr - I've used the base of this project https://github.com/cvasilak/kafka-connect-transforms to change the class/package to be specific to org.example.kafka.connect.transforms/Namespacefy, and then added the built jar to my classpath, but am still seeing the following on validation {"error_code":400,"message":"Connector configuration is invalid and contains the following 2 error(s):\nInvalid value org.example.kafka.connect.transforms.Namespacefy for configuration transforms.AddNamespace.type: Class org.example.kafka.connect.transforms.Namespacefy could not be found.\nInvalid value null for configuration transforms.AddNamespace.type: Not a Transformation\nYou can also find the above list of errors at the endpoint /{connectorType}/config/validate"} - this must mean the jar didn't get added correctly? Or is there an issue with the version I'm using 3.3.0 potentially?

Config has the following:
...
"transforms":"AddNamespace",
"transforms.AddNamespace.type":"org.example.kafka.connect.transforms.Namespacefy",
"transforms.AddNamespace.timestamp.record.namespace":"example"
}

Thanks in advance for any insight.

[kalbmj]

Looks like we're up and running with the suggestions from above @vikas-tikoo-zefr . Thanks again.

[bhumi1102]

I run into this issue as well (and was able to use the workaround, once I found this issue). It would be great if this basic use case can be supported by default in JDBC Source Connector.

[mattferroni]

Thanks @vikas-tikoo-zefr , you saved me as well! :)

[aoboturov]

@vikas-tikoo-zefr what is the location where the jar should be placed? Is it CONNECT_PLUGIN_PATH?

[vikas-tikoo-zefr]

@aoboturov yup. Thats been our setup as well.

[wicknicks]

Closing this since the workaround has solved the problem for multiple people.

[grzi]

Note :

The real workaround here is to use the native SetSchemaMetadata SMT.
In the schema.name field, just pout "your.package.name.ClassName"

The generated Avro Schema will use "your.package.name" as the namespace, and "ClassName" as the name.

Example :

"transforms":"AddNamespace", 
"transforms.AddNamespace.type":"org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
"transforms.AddNamespace.schema.name": "your.package.name.ClassName"

[helpermethod]

@zthulj I'm importing hundreds of tables. Isn't there a way to add a common namespace without specifying the tablename/classname?

With the JDBC source connector that should be a pretty common case.

[codependent]

I agree with @helpermethod in which there should be an out of the box way to just insert the namespace.

[mlundela]

@helpermethod @codependent The work around mention in this thread solves this, although it took me some time to get it right. I packaged the custom plugin mention here in a fat jar and added it to a kafka connect plugin directory, and then used the transform mention here with a minor modification:

```
...
"transforms":"AddNamespace",
"transforms.AddNamespace.type":"my.fat.jar.Namespacefy",
"transforms.AddNamespace.record.namespace":"{insert namespace here}"
}

[mbieser]

Supported natively as @grzi mentioned.
https://issues.apache.org/jira/browse/KAFKA-7883

[daniel-shuy]

As @helpermethod has mentioned, the workaround only works when importing a single table with a single connector, it is not a viable workaround when importing multiple tables with a single connector, and creating hundreds of connectors is not practical.