Hi all, we are using FlinkSQL with the Confluent Avro format. We are trying to find a way to make our create table query register a schema either with a given namespace or just use a schema we manually registered and skip the auto creation. For the second option, we tried to provide the following configs, but they don't seem to do what we expect - using the latest registered schema. Are we missing something? Also, any other ideas about how to force the namespace name will be helpful.

'value.avro-confluent.properties.use.latest.version' = 'true',
'value.avro-confluent.properties.auto.register.schemas' = 'false'

To expand on this a bit. What we're doing is this... We have a FlinkSQL job for which we create the table and then insert into it. Flink registers a schema with a generic namespace and name

org.apache.flink.avro.generated.record

. But when we then use these messages in the context of a Scala job, we want to use a specific record. So we deserialize it with a specific record. But because the specific record is set to true, the deserializer tries to load the ``org.apache.flink.avro.generated.record` class using a classloader and which doesn't exist. There is currently no way to specify the schema you want to use (or specify namespace/name) for the schema that Flink SQL generates. So we thought we can use the above options to force flink SQL to use an existing custom schema which we manually registered. The only diff in custom schema is namespace and name. But those options don't seem to work the way we assume they should. What is the best way to solve this issue, where we use Flink SQL to write to a topic (and it generates a generic avro schema), but then read it as a specific one?

Have you been able to solve this? I ran into the same issue and it’s very annoying. One potential solution I thought of is to convert from your initial table to a data stream. Then in that datastream, serialize the data manually with your confluent schema registry library. Then convert that datastream back to a table but with

format=raw

. That way you’re not relying on the table api to handle the serialization, just producing the already serialized data.

@Brian Koehler We're using SpecificData.deepCopy to do this manually. Basically, we grab the generic and then convert to specific.

For SpecificData.deepCopy, are you implementing that directly in Scala? I’m restricted to Python for my current project and not sure if I’m able to do that. Also, we have to use AWS Kinesis Data Analytics so 1.18 is a loooong way off

SpecificData.deepCopy is a method in an the avro library for Java

I see…doesn’t look like it’s available in PyFlink, at least in 15.4