Skip to content

[FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry #26662

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MartijnVisser wants to merge 4 commits into
base: master
Choose a base branch
from
+1,128 −202
Open

[FLINK-33045] Make it possible to disable auto-registering schema in Schema Registry #26662

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9f1c1cc
[FLINK-33045][formats/avro-confluent] Bump Confluent Schema Registry …
MartijnVisser Jul 3, 2026
f129d9d
[FLINK-33045][formats/avro-confluent] Add auto.register.schemas optio…
MartijnVisser Jul 3, 2026
4c65cd3
[FLINK-33045][e2e] Replace disabled bash-based schema registry test w…
MartijnVisser Jul 3, 2026
df1d162
[FLINK-33045][docs] Document schema auto-registration behavior for av…
MartijnVisser Jul 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions docs/content.zh/docs/connectors/table/formats/avro-confluent.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,13 @@ Format 参数
<td>String</td>
<td>Specify what format to use, here should be <code>'avro-confluent'</code>.</td>
</tr>
<tr>
<td><h5>avro-confluent.auto.register.schemas</h5></td>
<td>optional</td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
<td>Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to <code>false</code>, an identical schema must have been registered under the subject outside of Flink before it can be used; only its id is looked up and the job fails if it cannot be found. Schema registration only happens when writing data; reading always looks up the schema by the id embedded in each record.</td>
</tr>
<tr>
<td><h5>avro-confluent.basic-auth.credentials-source</h5></td>
<td>optional</td>
Expand Down
7 changes: 7 additions & 0 deletions docs/content.zh/docs/connectors/table/formats/debezium.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,13 @@ Flink 提供了 `debezium-avro-confluent` 和 `debezium-json` 两种 format 来
<td>String</td>
<td>Specify what format to use, here should be <code>'debezium-avro-confluent'</code>.</td>
</tr>
<tr>
<td><h5>debezium-avro-confluent.auto.register.schemas</h5></td>
<td>optional</td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
<td>Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to <code>false</code>, an identical schema must have been registered under the subject outside of Flink before it can be used; only its id is looked up and the job fails if it cannot be found. Schema registration only happens when writing data; reading always looks up the schema by the id embedded in each record.</td>
</tr>
<tr>
<td><h5>debezium-avro-confluent.basic-auth.credentials-source</h5></td>
<td>optional</td>
Expand Down
10 changes: 10 additions & 0 deletions docs/content/docs/connectors/table/formats/avro-confluent.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ When reading (deserializing) a record with this format the Avro writer schema is

When writing (serializing) a record with this format the Avro schema is inferred from the table schema and used to retrieve a schema id to be encoded with the data. The lookup is performed with in the configured Confluent Schema Registry under the [subject](https://docs.confluent.io/current/schema-registry/index.html#schemas-subjects-and-topics) given in `avro-confluent.subject`.

By default, the schema is automatically registered under the subject if it is not present yet (`'avro-confluent.auto.register.schemas' = 'true'`). Schema registration only happens when writing (serializing) data; reading always fetches the writer schema by the id encoded in each record. Automatic registration can be disabled (`'avro-confluent.auto.register.schemas' = 'false'`), for example when schemas are managed outside of Flink and write access to the Schema Registry is restricted. In that case a schema identical to the one used by the format must already be registered under the subject, otherwise the job fails with a "Schema not found" error. Since Flink infers the Avro schema from the table schema (using the record name `org.apache.flink.avro.generated.record`), it is recommended to provide the expected schema explicitly via `avro-confluent.schema` when schemas are managed externally. Note that the Schema Registry client caches failed lookups; registering a missing schema only takes effect after the cache entry expires or the job is restarted.

The Avro Schema Registry format can only be used in conjunction with the [Apache Kafka SQL connector]({{< ref "docs/connectors/table/kafka" >}}) or the [Upsert Kafka SQL Connector]({{< ref "docs/connectors/table/upsert-kafka" >}}).

Dependencies
Expand Down Expand Up @@ -191,6 +193,14 @@ Format Options
<td>String</td>
<td>Specify what format to use, here should be <code>'avro-confluent'</code>.</td>
</tr>
<tr>
<td><h5>avro-confluent.auto.register.schemas</h5></td>
<td>optional</td>
<td>yes</td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
<td>Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to <code>false</code>, an identical schema must have been registered under the subject outside of Flink before it can be used; only its id is looked up and the job fails if it cannot be found. Schema registration only happens when writing data; reading always looks up the schema by the id embedded in each record.</td>
</tr>
<tr>
<td><h5>avro-confluent.basic-auth.credentials-source</h5></td>
<td>optional</td>
Expand Down
7 changes: 7 additions & 0 deletions docs/content/docs/connectors/table/formats/debezium.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -282,6 +282,13 @@ Use format `debezium-avro-confluent` to interpret Debezium Avro messages and for
<td>String</td>
<td>Specify what format to use, here should be <code>'debezium-avro-confluent'</code>.</td>
</tr>
<tr>
<td><h5>debezium-avro-confluent.auto.register.schemas</h5></td>
<td>optional</td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
<td>Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to <code>false</code>, an identical schema must have been registered under the subject outside of Flink before it can be used; only its id is looked up and the job fails if it cannot be found. Schema registration only happens when writing data; reading always looks up the schema by the id embedded in each record.</td>
</tr>
<tr>
<td><h5>debezium-avro-confluent.basic-auth.credentials-source</h5></td>
<td>optional</td>
Expand Down
157 changes: 108 additions & 49 deletions flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,101 +43,160 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<artifactId>flink-table-common</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>3.0.0-1.17</version>
<artifactId>flink-table-api-java-bridge</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<!-- Make sure that Shaded Guava matches the one used in the flink-connector-kafka,
or remove when FLINK-32462 is resolved -->

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-guava</artifactId>
<version>30.1.1-jre-16.1</version>
<artifactId>flink-end-to-end-tests-common</artifactId>
<version>${project.version}</version>
</dependency>

<!-- This enables the WebUI during tests. -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-avro</artifactId>
<artifactId>flink-runtime-web</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>

<!-- The following dependencies are for connector/format sql-jars that
we copy using the maven-dependency-plugin. When extending the test
to cover more connectors/formats, add a dependency here and an entry
to the dependency-plugin configuration below.
This ensures that all modules we actually need (as defined by the
dependency-plugin configuration) are built before this module. -->
<dependency>
<!-- Used by maven-dependency-plugin -->
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-connector-kafka</artifactId>
<version>5.0.0-2.2</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-avro-confluent-registry</artifactId>
<artifactId>flink-sql-avro-confluent-registry</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>4.0.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-kafka</artifactId>
<scope>test</scope>
</dependency>

<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>7.9.8</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>1.11.5</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<scope>test</scope>
</dependency>

</dependencies>

<dependencyManagement>
<dependencies>
<dependency>
<!-- Pick an arbitrary version here to satisfy the enforcer-plugin,
as we neither access nor package the kafka dependencies -->
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>4.0.0</version>
</dependency>
</dependencies>
</dependencyManagement>

<build>
<plugins>
<!-- Build toolbox jar. -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<id>TestAvroConsumerConfluent</id>
<!-- Use a special execution id to be ignored by license/optional checks -->
<id>e2e-dependencies</id>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>TestAvroConsumerConfluent</finalName>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>org.apache.flink.schema.registry.test.TestAvroConsumerConfluent</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.avro</groupId>
<artifactId>avro-maven-plugin</artifactId>
<version>${avro.version}</version>
<executions>
<execution>
<phase>generate-sources</phase>
<goals>
<goal>schema</goal>
</goals>
<configuration>
<sourceDirectory>${project.basedir}/src/main/resources/avro/</sourceDirectory>
<outputDirectory>${project.basedir}/target/generated-sources/</outputDirectory>
<fieldVisibility>PRIVATE</fieldVisibility>
<includes>
<include>**/*.avsc</include>
</includes>
<finalName>SqlToolbox</finalName>
</configuration>
</execution>
</executions>
</plugin>

<!-- Copy SQL jars into dedicated "sql-jars" directory. -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-enforcer-plugin</artifactId>
<artifactId>maven-dependency-plugin</artifactId>
<executions>
<execution>
<id>dependency-convergence</id>
<id>copy</id>
<phase>package</phase>
<goals>
<goal>enforce</goal>
<goal>copy</goal>
</goals>
<configuration>
<skip>true</skip>
<outputDirectory>${project.build.directory}/sql-jars</outputDirectory>
<!-- List of currently provided SQL jars.
When extending this list please also add a dependency
for the respective module. -->
<artifactItems>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-connector-kafka</artifactId>
<version>5.0.0-2.2</version>
<type>jar</type>
</artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-test-utils</artifactId>
<version>${project.version}</version>
<type>jar</type>
</artifactItem>
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-avro-confluent-registry</artifactId>
<version>${project.version}</version>
<type>jar</type>
</artifactItem>
</artifactItems>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<configuration>
<excludes>**/example/avro/*</excludes>
</configuration>
</plugin>
</plugins>
</build>
</project>
21 changes: 11 additions & 10 deletions ...egistry/src/main/resources/avro/user.avsc → ...src/main/resources/avro/input-record.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,14 @@
* limitations under the License.
*/

{"namespace": "example.avro",
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string", "default": ""},
{"name": "favoriteNumber", "type": "string", "default": ""},
{"name": "favoriteColor", "type": "string", "default": ""},
{"name": "eventType","type": {"name": "EventType","type": "enum", "symbols": ["meeting"] }}
]
}
{
"namespace": "org.apache.flink.avro.generated",
"type": "record",
"name": "record",
"fields": [
{"name": "name", "type": ["null", "string"], "default": null},
Comment thread
MartijnVisser marked this conversation as resolved.
{"name": "favoriteNumber", "type": ["null", "string"], "default": null},
{"name": "favoriteColor", "type": ["null", "string"], "default": null},
{"name": "eventType", "type": ["null", "string"], "default": null}
]
}
28 changes: 28 additions & 0 deletions ...d-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/output-record.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

{
"namespace": "org.apache.flink.avro.generated",
"type": "record",
"name": "record",
"fields": [
{"name": "name", "type": ["null", "string"], "default": null},
{"name": "favoriteNumber", "type": ["null", "string"], "default": null},
{"name": "favoriteColor", "type": ["null", "string"], "default": null},
{"name": "eventType", "type": ["null", "string"], "default": null}
]
}
Loading