Kafka-connect-jdbc: Is there any way to get a nullable nested Avro record into Postgres?

[ceejsmith]

We are currently ingesting several Kafka topics with Avro schemas into Postgres using this sink connector. The topics have nested records, which the Flatten transform happily deals with, e.g. turning a metadata field that is a nested record into database columns with names such as metadata.eventId, metadata.createdAt.

However, we are running into problems when the nested record is declared as nullable in the Avro schema. We would expect Flatten to create . separated columns in the same way, but make them nullable and insert null if the record is null for a given Kafka message.

However, it explodes with a NPE:
Caused by: java.lang.NullPointerException
at org.apache.kafka.connect.transforms.Flatten.buildWithSchema(Flatten.java:219)
at org.apache.kafka.connect.transforms.Flatten.buildWithSchema(Flatten.java:234)
at org.apache.kafka.connect.transforms.Flatten.applyWithSchema(Flatten.java:151)

From looking at the source code, this happens when the transform is being built from the Avro schema. The nested nullable record doesn't have its own schema. The code that explodes is record.schema().fields() and we suspect that the nested nullable record is returning null for the .schema() call.

It would be great to get some pointers on this. Otherwise, we will have to write our own custom transform to figure out what's going on (or run a debug build of the current Flatten transform locally).

Thank you.

[woodlee]

I ran into this today. After some searching I found this Kafka Connect JIRA issue (which is probably the right place for it): https://issues.apache.org/jira/browse/KAFKA-6605

Ever come up with a decent workaround?

[ceejsmith]

Yes, we came up with an acceptable workaround, although it doesn't really fit into the overall intention of Kafka Connect. We wrote a "whitelist transformer" that specifies the output schema, e.g. customer.customerId:STRING, payment.amount: INTEGER.

It ignores the input schema, since that will vary with nullable nested records. That's where it doesn't really fit. The input schema is supposed to be transformed, not ignored :-)

The problem is that if "customer" is a nullable nested record in the overall schema and the first record doesn't have a customer record, the schema for the first record doesn't contain any mention of "customer", i.e. it doesn't say "there could be a customer here". That means you can't create, say, a nullable column in a relational database for "customer.customerId", which you expect to see in subsequent items.

So we decided to presuppose knowledge of the schema and configure the transform accordingly.

[ceejsmith]

Would it be useful to open source our code?

[woodlee]

Hi @ceejsmith , yes, if it's not too much trouble I'd love to have the code when you get a chance!

[HungUnicorn]

The test shows it should work?
https://github.com/apache/kafka/pull/5706/files#diff-4c075cf87e5b93f1ccdc86c589659fd5R201

to assure .schema call is not null, we can also check .schema and .schema.fields is not null just like the null check here
https://github.com/apache/kafka/pull/5706/files#diff-1fb37d9ee407ef678b0168966f70432bR223

[HungUnicorn]

I adapt the change in that PR and this should be able to use as a custom transform. If missing any unit test please lemme know but the unit test shows it should work(will test against my staging kafka-connect after days)
https://github.com/HungUnicorn/kafka-connect-transform-flatten-avro

[gharris1727]

It appears that this issue has been resolved in AK, and has been released in 2.4.0 and 2.3.1.

[HungUnicorn]

@gharris1727 can you please link the place where the code change, or a merge request?

[HungUnicorn]

ok found them
https://github.com/apache/kafka/pull/5705
https://issues.apache.org/jira/browse/KAFKA-6605