Hello, I am trying to create a table from Kafka relying on apicurio registry using the confluent compatibility offered by apicurio.

) WITH (
    'connector' = 'kafka',
    'property-version' = 'universal',
    'properties.bootstrap.servers' = 'host.docker.internal:29092',
    'scan.startup.mode' = 'earliest-offset',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.url' = '<http://host.docker.internal:8765/apis/ccompat/v6>', -- apicurio
    'topic' = 'my-topic',
    'properties.group.id' = 'mmy-consumer-group'
);

Here is the error

[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.avro.registry.confluent.shaded.io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: No content with id/hash 'contentId-0' was found.; error code: 40403

The endpoint runs fine

curl <http://localhost:8765/apis/ccompat/v6/schemas/ids/1>                                                                                                                                                                   

{"schema":"{\"type\":\"record\",\
...
}

But it seems that Flink isn't able to read the bytes containing the schema ID. It keeps reading

0

instead of

1, 2, ...

Has anyone ever faced this issue, and is there a config to add to tell Flink to look at the first 8 bytes for the artefact ID?

Does the compatibility mode of Apicurio also extend to the AVRO Confluent format? Or is it only the Schema Registry that's made compatible?

I think it's just the schema registry! Not sure how I can check it to be honest

Hi @Abdelhakim Bendjabeur? Have you been able to use confluent avro on flink without apicurio?

Hi Reme, no I haven't tried it, we only use apicurio