sqlLines) throws Exception {
+ LOG.info("Submitting SQL statements: {}", String.join("\n", sqlLines));
+ FLINK.submitSQLJob(
+ new SQLJobSubmission.SQLJobSubmissionBuilder(sqlLines)
+ .addJars(
+ sqlConnectorUpsertTestJar,
+ sqlConnectorKafkaJar,
+ sqlAvroConfluentJar,
+ sqlToolBoxJar)
+ .build());
+ }
+}
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/SchemaLoader.java b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/SchemaLoader.java
new file mode 100644
index 00000000000000..d4a0023aa54317
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/SchemaLoader.java
@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+
+package org.apache.flink.schema.registry.test;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+
+/** Utility class for loading Avro schema files from resources. */
+public class SchemaLoader {
+
+ /**
+ * Load an Avro schema from a resource file.
+ *
+ * @param resourcePath Path to the .avsc file in resources
+ * @return Schema content as string
+ * @throws RuntimeException if the file cannot be loaded
+ */
+ public static String loadSchema(String resourcePath) {
+ try (InputStream inputStream =
+ SchemaLoader.class.getClassLoader().getResourceAsStream(resourcePath)) {
+ if (inputStream == null) {
+ throw new RuntimeException("Schema file not found: " + resourcePath);
+ }
+ return new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to load schema from: " + resourcePath, e);
+ }
+ }
+}
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/avro_confluent_missing_schema_e2e.sql b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/avro_confluent_missing_schema_e2e.sql
new file mode 100644
index 00000000000000..8c1678f2e64cf4
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/avro_confluent_missing_schema_e2e.sql
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+SET 'table.dml-sync' = 'true';
+
+CREATE TABLE avro_output_missing_schema (
+ name STRING,
+ favoriteNumber STRING,
+ favoriteColor STRING,
+ eventType STRING
+) WITH (
+ 'connector' = 'kafka',
+ 'topic' = '$OUTPUT_TOPIC',
+ 'properties.bootstrap.servers' = '$BOOTSTRAP_SERVERS',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'scan.bounded.mode' = 'latest-offset',
+ 'format' = 'avro-confluent',
+ 'avro-confluent.url' = '$SCHEMA_REGISTRY_URL',
+ 'avro-confluent.auto.register.schemas' = 'false'
+);
+
+INSERT INTO avro_output_missing_schema VALUES ('Grace', '1', 'black', 'INSERT');
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/avro_confluent_roundtrip_e2e.sql b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/avro_confluent_roundtrip_e2e.sql
new file mode 100644
index 00000000000000..de936e893aa6d9
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/avro_confluent_roundtrip_e2e.sql
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+SET 'table.dml-sync' = 'true';
+
+CREATE TABLE avro_input (
+ name STRING,
+ favoriteNumber STRING,
+ favoriteColor STRING,
+ eventType STRING
+) WITH (
+ 'connector' = 'kafka',
+ 'topic' = '$INPUT_TOPIC',
+ 'properties.bootstrap.servers' = '$BOOTSTRAP_SERVERS',
+ 'properties.group.id' = 'test-group-$INPUT_TOPIC',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'scan.bounded.mode' = 'latest-offset',
+ 'format' = 'avro-confluent',
+ 'avro-confluent.url' = '$SCHEMA_REGISTRY_URL',
+ 'avro-confluent.auto.register.schemas' = '$AUTO_REGISTER_SCHEMAS'
+);
+
+CREATE TABLE avro_output (
+ name STRING,
+ favoriteNumber STRING,
+ favoriteColor STRING,
+ eventType STRING
+) WITH (
+ 'connector' = 'kafka',
+ 'topic' = '$OUTPUT_TOPIC',
+ 'properties.bootstrap.servers' = '$BOOTSTRAP_SERVERS',
+ 'properties.group.id' = 'test-group-$OUTPUT_TOPIC',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'scan.bounded.mode' = 'latest-offset',
+ 'format' = 'avro-confluent',
+ 'avro-confluent.url' = '$SCHEMA_REGISTRY_URL',
+ 'avro-confluent.auto.register.schemas' = '$AUTO_REGISTER_SCHEMAS'
+);
+
+INSERT INTO avro_input VALUES
+ ('Alice', '42', 'blue', 'INSERT'),
+ ('Bob', '7', 'red', 'INSERT'),
+ ('Charlie', '73', 'green', 'INSERT');
+
+INSERT INTO avro_output
+SELECT * FROM avro_input;
diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/log4j2-test.properties b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/log4j2-test.properties
new file mode 100644
index 00000000000000..358fd81ef99786
--- /dev/null
+++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/resources/log4j2-test.properties
@@ -0,0 +1,34 @@
+################################################################################
+# 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.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+
+# It is recommended to uncomment these lines when enabling the logger. The below package used
+# by testcontainers is quite verbose
+#logger.yarn.name = org.testcontainers.shaded.com.github.dockerjava.core
+#logger.yarn.level = WARN
+#logger.yarn.appenderRef.console.ref = TestLogger
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 3b84d479836fbf..b2cb9bd0e82ab5 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -194,10 +194,6 @@ function run_group_3 {
# Miscellaneous
################################################################################
- # Disable this test as 2.0 broke the compatibility of kafka sink writer. We should consider migrate this test to flink-connector-kafka repo.
- # See FLINK-36268.
- # run_test "Avro Confluent Schema Registry nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_confluent_schema_registry.sh"
-
run_test "State TTL Heap backend end-to-end test" "$END_TO_END_DIR/test-scripts/test_stream_state_ttl.sh hashmap" "skip_check_exceptions"
run_test "State TTL RocksDb backend end-to-end test" "$END_TO_END_DIR/test-scripts/test_stream_state_ttl.sh rocks" "skip_check_exceptions"
diff --git a/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh b/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
deleted file mode 100755
index d653ef46bc578a..00000000000000
--- a/flink-end-to-end-tests/test-scripts/test_confluent_schema_registry.sh
+++ /dev/null
@@ -1,117 +0,0 @@
-#!/usr/bin/env bash
-################################################################################
-# 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.
-################################################################################
-
-set -Eeuo pipefail
-
-KAFKA_VERSION="3.2.3"
-CONFLUENT_VERSION="7.5.3"
-CONFLUENT_MAJOR_VERSION="7.5"
-# Check the Confluent Platform <> Apache Kafka compatibility matrix when updating KAFKA_VERSION
-KAFKA_SQL_VERSION="universal"
-
-source "$(dirname "$0")"/common.sh
-source "$(dirname "$0")"/kafka_sql_common.sh \
- $KAFKA_VERSION \
- $CONFLUENT_VERSION \
- $CONFLUENT_MAJOR_VERSION \
- $KAFKA_SQL_VERSION
-
-function verify_output {
- local expected=$(printf $1)
- local result=$(echo $2 | sed 's/ //g')
-
- if [[ "$result" != "$expected" ]]; then
- echo "Output from Flink program does not match expected output."
- echo -e "EXPECTED FOR KEY: --$expected--"
- echo -e "ACTUAL: --$result--"
- exit 1
- fi
-}
-
-function test_setup {
- start_kafka_cluster
- start_confluent_schema_registry
-}
-
-function test_cleanup {
- stop_confluent_schema_registry
- stop_kafka_cluster
-}
-
-on_exit test_cleanup
-
-function schema_registry_test {
- setup_kafka_dist
- setup_confluent_dist
-
- retry_times_with_backoff_and_cleanup 3 5 test_setup test_cleanup
-
- TEST_PROGRAM_JAR=${END_TO_END_DIR}/flink-confluent-schema-registry/target/TestAvroConsumerConfluent.jar
-
- INPUT_MESSAGE_1='{"name":"Alyssa","favoriteNumber":"250","favoriteColor":"green","eventType":"meeting"}'
- INPUT_MESSAGE_2='{"name":"Charlie","favoriteNumber":"10","favoriteColor":"blue","eventType":"meeting"}'
- INPUT_MESSAGE_3='{"name":"Ben","favoriteNumber":"7","favoriteColor":"red","eventType":"meeting"}'
- USER_SCHEMA='{"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"]}}]}'
-
- curl -X POST \
- ${SCHEMA_REGISTRY_URL}/subjects/users-value/versions \
- -H 'cache-control: no-cache' \
- -H 'content-type: application/vnd.schemaregistry.v1+json' \
- -d '{"schema": "{\"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\"] }}]}"}'
-
- echo "Sending messages to Kafka topic [test-avro-input] ..."
-
- send_messages_to_kafka_avro $INPUT_MESSAGE_1 test-avro-input $USER_SCHEMA
- send_messages_to_kafka_avro $INPUT_MESSAGE_2 test-avro-input $USER_SCHEMA
- send_messages_to_kafka_avro $INPUT_MESSAGE_3 test-avro-input $USER_SCHEMA
-
- start_cluster
-
- create_kafka_topic 1 1 test-string-out
- create_kafka_topic 1 1 test-avro-out
-
- # Read Avro message from [test-avro-input], check the schema and send message to [test-string-ou]
- $FLINK_DIR/bin/flink run -d $TEST_PROGRAM_JAR \
- --input-topic test-avro-input --output-string-topic test-string-out --output-avro-topic test-avro-out --output-subject test-output-subject \
- --bootstrap.servers localhost:9092 --group.id myconsumer --auto.offset.reset earliest \
- --schema-registry-url ${SCHEMA_REGISTRY_URL}
-
- echo "Reading messages from Kafka topic [test-string-out] ..."
-
- KEY_1_STRING_MSGS=$(read_messages_from_kafka 3 test-string-out Alyssa_consumer | grep Alyssa)
- KEY_2_STRING_MSGS=$(read_messages_from_kafka 3 test-string-out Charlie_consumer | grep Charlie)
- KEY_3_STRING_MSGS=$(read_messages_from_kafka 3 test-string-out Ben_consumer | grep Ben)
-
- ## Verifying STRING output with actual message
- verify_output $INPUT_MESSAGE_1 "$KEY_1_STRING_MSGS"
- verify_output $INPUT_MESSAGE_2 "$KEY_2_STRING_MSGS"
- verify_output $INPUT_MESSAGE_3 "$KEY_3_STRING_MSGS"
-
- KEY_1_AVRO_MSGS=$(read_messages_from_kafka_avro 3 test-avro-out $USER_SCHEMA Alyssa_consumer_1 | grep Alyssa)
- KEY_2_AVRO_MSGS=$(read_messages_from_kafka_avro 3 test-avro-out $USER_SCHEMA Charlie_consumer_1 | grep Charlie)
- KEY_3_AVRO_MSGS=$(read_messages_from_kafka_avro 3 test-avro-out $USER_SCHEMA Ben_consumer_1 | grep Ben)
-
- ## Verifying AVRO output with actual message
- verify_output $INPUT_MESSAGE_1 "$KEY_1_AVRO_MSGS"
- verify_output $INPUT_MESSAGE_2 "$KEY_2_AVRO_MSGS"
- verify_output $INPUT_MESSAGE_3 "$KEY_3_AVRO_MSGS"
-}
-
-run_test_with_timeout 900 schema_registry_test
-
From df1d162190270a28ce803c651a5a05855e76310c Mon Sep 17 00:00:00 2001
From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com>
Date: Fri, 3 Jul 2026 23:15:42 +0200
Subject: [PATCH 4/4] [FLINK-33045][docs] Document schema auto-registration
behavior for avro-confluent formats
---
.../docs/connectors/table/formats/avro-confluent.md | 7 +++++++
.../docs/connectors/table/formats/debezium.md | 7 +++++++
.../docs/connectors/table/formats/avro-confluent.md | 10 ++++++++++
docs/content/docs/connectors/table/formats/debezium.md | 7 +++++++
4 files changed, 31 insertions(+)
diff --git a/docs/content.zh/docs/connectors/table/formats/avro-confluent.md b/docs/content.zh/docs/connectors/table/formats/avro-confluent.md
index 02b71fab66d261..6caa2fdd9ea9ed 100644
--- a/docs/content.zh/docs/connectors/table/formats/avro-confluent.md
+++ b/docs/content.zh/docs/connectors/table/formats/avro-confluent.md
@@ -196,6 +196,13 @@ Format 参数
| String |
Specify what format to use, here should be 'avro-confluent'. |
+
+ avro-confluent.auto.register.schemas |
+ optional |
+ true |
+ Boolean |
+ Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to false, 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. |
+
avro-confluent.basic-auth.credentials-source |
optional |
diff --git a/docs/content.zh/docs/connectors/table/formats/debezium.md b/docs/content.zh/docs/connectors/table/formats/debezium.md
index a0b2f8d7b7a295..0991ceba1e3be2 100644
--- a/docs/content.zh/docs/connectors/table/formats/debezium.md
+++ b/docs/content.zh/docs/connectors/table/formats/debezium.md
@@ -292,6 +292,13 @@ Flink 提供了 `debezium-avro-confluent` 和 `debezium-json` 两种 format 来
String |
Specify what format to use, here should be 'debezium-avro-confluent'. |
+
+ debezium-avro-confluent.auto.register.schemas |
+ optional |
+ true |
+ Boolean |
+ Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to false, 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. |
+
debezium-avro-confluent.basic-auth.credentials-source |
optional |
diff --git a/docs/content/docs/connectors/table/formats/avro-confluent.md b/docs/content/docs/connectors/table/formats/avro-confluent.md
index c2fb1bc4717867..e8f4968801bd7e 100644
--- a/docs/content/docs/connectors/table/formats/avro-confluent.md
+++ b/docs/content/docs/connectors/table/formats/avro-confluent.md
@@ -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
@@ -191,6 +193,14 @@ Format Options
String |
Specify what format to use, here should be 'avro-confluent'. |
+
+ avro-confluent.auto.register.schemas |
+ optional |
+ yes |
+ true |
+ Boolean |
+ Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to false, 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. |
+
avro-confluent.basic-auth.credentials-source |
optional |
diff --git a/docs/content/docs/connectors/table/formats/debezium.md b/docs/content/docs/connectors/table/formats/debezium.md
index 54f36448887e21..8623a265d3c5ed 100644
--- a/docs/content/docs/connectors/table/formats/debezium.md
+++ b/docs/content/docs/connectors/table/formats/debezium.md
@@ -282,6 +282,13 @@ Use format `debezium-avro-confluent` to interpret Debezium Avro messages and for
String |
Specify what format to use, here should be 'debezium-avro-confluent'. |
+
+ debezium-avro-confluent.auto.register.schemas |
+ optional |
+ true |
+ Boolean |
+ Whether to automatically register the schema in the Confluent Schema Registry during serialization. When set to false, 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. |
+
debezium-avro-confluent.basic-auth.credentials-source |
optional |