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 diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml b/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml index e115b997f40f28..98f215063f20a7 100644 --- a/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml +++ b/flink-end-to-end-tests/flink-confluent-schema-registry/pom.xml @@ -43,101 +43,160 @@ under the License. org.apache.flink - flink-streaming-java + flink-table-common ${project.version} provided - + org.apache.flink - flink-connector-kafka - 3.0.0-1.17 + flink-table-api-java-bridge + ${project.version} + provided - + org.apache.flink - flink-shaded-guava - 30.1.1-jre-16.1 + flink-end-to-end-tests-common + ${project.version} + + org.apache.flink - flink-avro + flink-runtime-web ${project.version} + test + + + + + + org.apache.flink + flink-sql-connector-kafka + 5.0.0-2.2 + provided + org.apache.flink - flink-avro-confluent-registry + flink-sql-avro-confluent-registry ${project.version} + provided + + + + org.apache.kafka + kafka-clients + 4.0.0 + test + + + org.testcontainers + testcontainers-kafka + test + + + + io.confluent + kafka-avro-serializer + 7.9.8 + test + + + + org.apache.avro + avro + 1.11.5 + test + + + + commons-codec + commons-codec + test + + + + + + org.apache.kafka + kafka-clients + 4.0.0 + + + + + org.apache.maven.plugins maven-shade-plugin - TestAvroConsumerConfluent + + e2e-dependencies package shade - TestAvroConsumerConfluent - - - org.apache.flink.schema.registry.test.TestAvroConsumerConfluent - - - - - - - - org.apache.avro - avro-maven-plugin - ${avro.version} - - - generate-sources - - schema - - - ${project.basedir}/src/main/resources/avro/ - ${project.basedir}/target/generated-sources/ - PRIVATE - - **/*.avsc - + SqlToolbox + + org.apache.maven.plugins - maven-enforcer-plugin + maven-dependency-plugin - dependency-convergence + copy + package - enforce + copy - true + ${project.build.directory}/sql-jars + + + + org.apache.flink + flink-sql-connector-kafka + 5.0.0-2.2 + jar + + + org.apache.flink + flink-test-utils + ${project.version} + jar + + + org.apache.flink + flink-sql-avro-confluent-registry + ${project.version} + jar + + - - org.apache.maven.plugins - maven-checkstyle-plugin - - **/example/avro/* - - diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/user.avsc b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/input-record.avsc similarity index 66% rename from flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/user.avsc rename to flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/input-record.avsc index aca9b83f9c2ced..bb275a71b01ce4 100644 --- a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/user.avsc +++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/input-record.avsc @@ -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}, + {"name": "favoriteNumber", "type": ["null", "string"], "default": null}, + {"name": "favoriteColor", "type": ["null", "string"], "default": null}, + {"name": "eventType", "type": ["null", "string"], "default": null} + ] +} diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/output-record.avsc b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/output-record.avsc new file mode 100644 index 00000000000000..bb275a71b01ce4 --- /dev/null +++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/main/resources/avro/output-record.avsc @@ -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} + ] +} diff --git a/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java new file mode 100644 index 00000000000000..72bb396873a5f4 --- /dev/null +++ b/flink-end-to-end-tests/flink-confluent-schema-registry/src/test/java/org/apache/flink/schema/registry/test/AvroConfluentITCase.java @@ -0,0 +1,439 @@ +/* + * 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 org.apache.flink.api.common.JobStatus; +import org.apache.flink.connector.testframe.container.FlinkContainers; +import org.apache.flink.connector.testframe.container.FlinkContainersSettings; +import org.apache.flink.connector.testframe.container.TestcontainersSettings; +import org.apache.flink.core.testutils.CommonTestUtils; +import org.apache.flink.runtime.client.JobStatusMessage; +import org.apache.flink.test.resources.ResourceTestUtils; +import org.apache.flink.test.util.SQLJobSubmission; +import org.apache.flink.util.DockerImageVersions; +import org.apache.flink.util.jackson.JacksonMapperFactory; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +import org.apache.avro.generic.GenericRecord; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.kafka.ConfluentKafkaContainer; +import org.testcontainers.utility.DockerImageName; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.net.URI; +import java.net.URL; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.TimeoutException; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** E2E Test for Avro-Confluent integration. */ +@Testcontainers +public class AvroConfluentITCase { + + private static final Logger LOG = LoggerFactory.getLogger(AvroConfluentITCase.class); + + private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka"; + private static final String INTER_CONTAINER_KAFKA_BOOTSTRAP = + INTER_CONTAINER_KAFKA_ALIAS + ":19092"; + private static final String INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS = "schema-registry"; + private static final String INTER_CONTAINER_SCHEMA_REGISTRY_URL = + "http://" + INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS + ":8081"; + + private static final String TOPIC = "test-avro-input"; + private static final String RESULT_TOPIC = "test-avro-output"; + private static final String MANUAL_TOPIC = "test-avro-input-manual"; + private static final String MANUAL_RESULT_TOPIC = "test-avro-output-manual"; + private static final String MISSING_SCHEMA_TOPIC = "test-avro-output-missing-schema"; + + private static final String INPUT_SCHEMA = SchemaLoader.loadSchema("avro/input-record.avsc"); + private static final String OUTPUT_SCHEMA = SchemaLoader.loadSchema("avro/output-record.avsc"); + + private static final Path sqlToolBoxJar = ResourceTestUtils.getResource(".*/SqlToolbox\\.jar"); + + private final Path sqlConnectorKafkaJar = ResourceTestUtils.getResource(".*kafka.*\\.jar"); + private final Path sqlConnectorUpsertTestJar = + ResourceTestUtils.getResource(".*flink-test-utils.*\\.jar"); + private final Path sqlAvroConfluentJar = + ResourceTestUtils.getResource(".*avro-confluent.*\\.jar"); + + private static final Network NETWORK = Network.newNetwork(); + + private static final HttpClient HTTP_CLIENT = HttpClient.newHttpClient(); + private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper(); + + @Container + public static final ConfluentKafkaContainer KAFKA = + new ConfluentKafkaContainer(DockerImageName.parse(DockerImageVersions.KAFKA)) + .withNetwork(NETWORK) + .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS) + .withListener(INTER_CONTAINER_KAFKA_BOOTSTRAP); + + @Container + private static final GenericContainer SCHEMA_REGISTRY = + new GenericContainer<>(DockerImageName.parse(DockerImageVersions.SCHEMA_REGISTRY)) + .withNetwork(NETWORK) + .withNetworkAliases(INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS) + .withExposedPorts(8081) + .withEnv("SCHEMA_REGISTRY_HOST_NAME", INTER_CONTAINER_SCHEMA_REGISTRY_ALIAS) + .withEnv( + "SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", + INTER_CONTAINER_KAFKA_BOOTSTRAP) + .dependsOn(KAFKA); + + @RegisterExtension + private static final FlinkContainers FLINK = + FlinkContainers.builder() + .withFlinkContainersSettings( + FlinkContainersSettings.builder().numTaskManagers(1).build()) + .withTestcontainersSettings( + TestcontainersSettings.builder() + .network(NETWORK) + .logger(LOG) + .dependsOn(KAFKA) + .build()) + .build(); + + @BeforeEach + public void before() throws Exception { + // Create topics using external bootstrap servers since we're outside the Docker network + Properties props = new Properties(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers()); + try (Admin admin = Admin.create(props)) { + List topics = + Arrays.asList( + new NewTopic(TOPIC, 1, (short) 1), + new NewTopic(RESULT_TOPIC, 1, (short) 1), + new NewTopic(MANUAL_TOPIC, 1, (short) 1), + new NewTopic(MANUAL_RESULT_TOPIC, 1, (short) 1), + new NewTopic(MISSING_SCHEMA_TOPIC, 1, (short) 1)); + admin.createTopics(topics); + + // Poll for topic creation with timeout + try { + CommonTestUtils.waitUntilIgnoringExceptions( + () -> { + try { + Set existingTopics = admin.listTopics().names().get(); + return topics.stream() + .allMatch(topic -> existingTopics.contains(topic.name())); + } catch (Exception e) { + LOG.warn("Exception while checking topic creation", e); + return false; + } + }, + Duration.ofSeconds(30), + Duration.ofMillis(100), + "Topics were not created in time"); + } catch (TimeoutException | InterruptedException e) { + throw new RuntimeException("Failed to wait for topic creation", e); + } + + LOG.info("Topics created successfully"); + } + } + + @Test + public void testAvroConfluentIntegrationWithAutoRegister() throws Exception { + List sqlLines = + initializeSqlLines( + "avro_confluent_roundtrip_e2e.sql", + generateRoundtripReplaceVars(TOPIC, RESULT_TOPIC, true)); + + executeSql(sqlLines); + + verifyNumberOfResultRecords(RESULT_TOPIC, 3); + } + + @Test + public void testAvroConfluentIntegrationWithManualRegister() throws Exception { + // Manually register schemas before creating tables + registerSchema(MANUAL_TOPIC + "-value", INPUT_SCHEMA); + registerSchema(MANUAL_RESULT_TOPIC + "-value", OUTPUT_SCHEMA); + + List sqlLines = + initializeSqlLines( + "avro_confluent_roundtrip_e2e.sql", + generateRoundtripReplaceVars(MANUAL_TOPIC, MANUAL_RESULT_TOPIC, false)); + + executeSql(sqlLines); + + verifyNumberOfResultRecords(MANUAL_RESULT_TOPIC, 3); + + // The job must not have registered any new schema version on top of the + // manually registered ones + assertThat(getSubjectVersions(MANUAL_TOPIC + "-value")).containsExactly(1); + assertThat(getSubjectVersions(MANUAL_RESULT_TOPIC + "-value")).containsExactly(1); + } + + @Test + public void testWritingFailsWhenAutoRegisterDisabledAndSchemaMissing() throws Exception { + // No schema is registered for MISSING_SCHEMA_TOPIC, so writing with + // auto.register.schemas=false must fail instead of registering the schema + Map vars = new HashMap<>(); + vars.put("$OUTPUT_TOPIC", MISSING_SCHEMA_TOPIC); + vars.put("$BOOTSTRAP_SERVERS", INTER_CONTAINER_KAFKA_BOOTSTRAP); + vars.put("$SCHEMA_REGISTRY_URL", INTER_CONTAINER_SCHEMA_REGISTRY_URL); + List sqlLines = initializeSqlLines("avro_confluent_missing_schema_e2e.sql", vars); + + // The SQL client reads the script from stdin and always exits successfully, even if + // a statement fails; the failure must be observed through the job status instead. + // 'table.dml-sync' guarantees the job has reached a terminal state afterwards. + executeSql(sqlLines); + + JobStatusMessage insertJob = + FLINK.getRestClusterClient().listJobs().get().stream() + .filter(job -> job.getJobName().contains("avro_output_missing_schema")) + .findAny() + .orElseThrow( + () -> + new AssertionError( + "Could not find the job writing to the missing-schema table")); + assertThat(insertJob.getJobState()).isEqualTo(JobStatus.FAILED); + + // The failed job must not have registered the schema either + assertThat(subjectExists(MISSING_SCHEMA_TOPIC + "-value")).isFalse(); + } + + private static Map generateRoundtripReplaceVars( + String inputTopic, String outputTopic, boolean autoRegisterSchemas) { + Map vars = new HashMap<>(); + vars.put("$INPUT_TOPIC", inputTopic); + vars.put("$OUTPUT_TOPIC", outputTopic); + vars.put("$BOOTSTRAP_SERVERS", INTER_CONTAINER_KAFKA_BOOTSTRAP); + vars.put("$SCHEMA_REGISTRY_URL", INTER_CONTAINER_SCHEMA_REGISTRY_URL); + vars.put("$AUTO_REGISTER_SCHEMAS", String.valueOf(autoRegisterSchemas)); + return vars; + } + + private static List initializeSqlLines(String sqlPath, Map vars) + throws IOException { + URL url = AvroConfluentITCase.class.getClassLoader().getResource(sqlPath); + if (url == null) { + throw new FileNotFoundException(sqlPath); + } + List lines = Files.readAllLines(new File(url.getFile()).toPath()); + List result = new ArrayList<>(); + for (String line : lines) { + for (Map.Entry var : vars.entrySet()) { + line = line.replace(var.getKey(), var.getValue()); + } + result.add(line); + } + return result; + } + + private static String getSchemaRegistryUrl() { + return "http://" + SCHEMA_REGISTRY.getHost() + ":" + SCHEMA_REGISTRY.getMappedPort(8081); + } + + private String registerSchema(String subject, String schema) { + try { + // Use Jackson to properly serialize the request body + Map requestBodyMap = new HashMap<>(); + requestBodyMap.put("schema", schema); + String requestBody = OBJECT_MAPPER.writeValueAsString(requestBodyMap); + + LOG.info("Registering schema for subject {} with schema: {}", subject, schema); + + HttpRequest request = + HttpRequest.newBuilder() + .uri( + URI.create( + getSchemaRegistryUrl() + + "/subjects/" + + subject + + "/versions")) + .header("Content-Type", "application/vnd.schemaregistry.v1+json") + .POST(HttpRequest.BodyPublishers.ofString(requestBody)) + .build(); + + HttpResponse response = + HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString()); + + if (response.statusCode() != 200) { + throw new RuntimeException( + "Failed to register schema for subject " + + subject + + ": " + + response.statusCode() + + " - " + + response.body()); + } + + LOG.info("Successfully registered schema for subject {}: {}", subject, response.body()); + return response.body(); + } catch (Exception e) { + throw new RuntimeException("Failed to register schema for subject " + subject, e); + } + } + + private List getSubjectVersions(String subject) { + try { + HttpRequest request = + HttpRequest.newBuilder() + .uri( + URI.create( + getSchemaRegistryUrl() + + "/subjects/" + + subject + + "/versions")) + .GET() + .build(); + HttpResponse response = + HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString()); + if (response.statusCode() != 200) { + throw new RuntimeException( + "Failed to get versions for subject " + + subject + + ": " + + response.statusCode() + + " - " + + response.body()); + } + return Arrays.asList(OBJECT_MAPPER.readValue(response.body(), Integer[].class)); + } catch (Exception e) { + throw new RuntimeException("Failed to get versions for subject " + subject, e); + } + } + + private boolean subjectExists(String subject) { + try { + HttpRequest request = + HttpRequest.newBuilder() + .uri( + URI.create( + getSchemaRegistryUrl() + + "/subjects/" + + subject + + "/versions")) + .GET() + .build(); + HttpResponse response = + HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString()); + return response.statusCode() == 200; + } catch (Exception e) { + throw new RuntimeException("Failed to check subject " + subject, e); + } + } + + private void verifyNumberOfResultRecords(String topicName, int expectedCount) throws Exception { + // Use external bootstrap servers since we're outside the Docker network + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers()); + props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-verification-group-" + topicName); + props.put( + ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.StringDeserializer"); + props.put( + ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + "io.confluent.kafka.serializers.KafkaAvroDeserializer"); + props.put("schema.registry.url", getSchemaRegistryUrl()); + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + + try (KafkaConsumer consumer = new KafkaConsumer<>(props)) { + consumer.subscribe(Collections.singletonList(topicName)); + List records = new ArrayList<>(); + + // Poll for records until we have enough or timeout + try { + CommonTestUtils.waitUntilIgnoringExceptions( + () -> { + try { + ConsumerRecords consumerRecords = + consumer.poll(Duration.ofMillis(100)); + for (ConsumerRecord record : + consumerRecords) { + records.add(record.value()); + } + return records.size() >= expectedCount; + } catch (Exception e) { + LOG.warn("Exception while polling records", e); + return false; + } + }, + Duration.ofSeconds(30), + Duration.ofMillis(100), + "Expected " + + expectedCount + + " records but only found " + + records.size() + + " in topic " + + topicName); + } catch (TimeoutException | InterruptedException e) { + throw new RuntimeException("Failed to wait for records in topic " + topicName, e); + } + + // Verify the exact number of records + assertEquals( + expectedCount, + records.size(), + "Expected exactly " + expectedCount + " records but found " + records.size()); + + // Log the records for debugging + for (GenericRecord record : records) { + LOG.info("Found record in {}: {}", topicName, record); + } + } + } + + private void executeSql(List 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 - diff --git a/flink-formats/flink-avro-confluent-registry/pom.xml b/flink-formats/flink-avro-confluent-registry/pom.xml index e8e68a15d6d99a..5fcc9eda3c3120 100644 --- a/flink-formats/flink-avro-confluent-registry/pom.xml +++ b/flink-formats/flink-avro-confluent-registry/pom.xml @@ -32,7 +32,7 @@ under the License. Flink : Formats : Avro confluent registry - 7.5.3 + 7.9.8 diff --git a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/AvroConfluentFormatOptions.java b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/AvroConfluentFormatOptions.java index 42e7353d009ba4..8a185d2e24eb0f 100644 --- a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/AvroConfluentFormatOptions.java +++ b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/AvroConfluentFormatOptions.java @@ -65,6 +65,17 @@ public class AvroConfluentFormatOptions { // Commonly used options maintained by Flink for convenience // -------------------------------------------------------------------------------------------- + public static final ConfigOption AUTO_REGISTER_SCHEMAS = + ConfigOptions.key("auto.register.schemas") + .booleanType() + .defaultValue(true) + .withDescription( + "Flag to indicate if the schema should be registered automatically during serialization. " + + "When set to false, an identical schema must have been registered in the Schema Registry " + + "outside of Flink before it can be used; only its id is looked up. " + + "Schema registration only happens when writing data; reading always looks up " + + "the schema by the id embedded in each record. The default value is true."); + public static final ConfigOption SSL_KEYSTORE_LOCATION = ConfigOptions.key("ssl.keystore.location") .stringType() diff --git a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/CachedSchemaCoderProvider.java b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/CachedSchemaCoderProvider.java index 53be09c7475a6c..d8a59846f69c3e 100644 --- a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/CachedSchemaCoderProvider.java +++ b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/CachedSchemaCoderProvider.java @@ -59,7 +59,8 @@ class CachedSchemaCoderProvider implements SchemaCoder.SchemaCoderProvider { public SchemaCoder get() { return new ConfluentSchemaRegistryCoder( this.subject, - new CachedSchemaRegistryClient(url, identityMapCapacity, registryConfigs)); + new CachedSchemaRegistryClient(url, identityMapCapacity, registryConfigs), + registryConfigs); } @Override diff --git a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java index 7be31453148356..3bf2a7a0aec9d0 100644 --- a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java +++ b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoder.java @@ -24,41 +24,87 @@ import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; import org.apache.avro.Schema; +import javax.annotation.Nullable; + import java.io.DataInputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.nio.ByteBuffer; +import java.util.Map; import static java.lang.String.format; /** Reads and Writes schema using Confluent Schema Registry protocol. */ public class ConfluentSchemaRegistryCoder implements SchemaCoder { + private static final String AUTO_REGISTER_SCHEMAS_CONFIG = "auto.register.schemas"; + private final SchemaRegistryClient schemaRegistryClient; + private final boolean autoRegisterSchemas; private String subject; private static final int CONFLUENT_MAGIC_BYTE = 0; /** - * Creates {@link SchemaCoder} that uses provided {@link SchemaRegistryClient} to connect to + * Creates {@link SchemaCoder} that uses provided {@link SchemaRegistryClient} to connect to the * schema registry. * - * @param schemaRegistryClient client to connect schema registry * @param subject subject of schema registry to produce + * @param schemaRegistryClient client to connect to the schema registry + * @param registryConfigs map with additional schema registry configs. Supports {@code + * auto.register.schemas} to control whether schemas are registered on write (default) or + * only looked up */ - public ConfluentSchemaRegistryCoder(String subject, SchemaRegistryClient schemaRegistryClient) { + public ConfluentSchemaRegistryCoder( + String subject, + SchemaRegistryClient schemaRegistryClient, + @Nullable Map registryConfigs) { this.schemaRegistryClient = schemaRegistryClient; this.subject = subject; + this.autoRegisterSchemas = parseAutoRegisterSchemas(registryConfigs); } /** - * Creates {@link SchemaCoder} that uses provided {@link SchemaRegistryClient} to connect to + * Creates {@link SchemaCoder} that uses provided {@link SchemaRegistryClient} to connect to the * schema registry. * - * @param schemaRegistryClient client to connect schema registry + * @param subject subject of schema registry to produce + * @param schemaRegistryClient client to connect to the schema registry + */ + public ConfluentSchemaRegistryCoder(String subject, SchemaRegistryClient schemaRegistryClient) { + this(subject, schemaRegistryClient, null); + } + + /** + * Creates {@link SchemaCoder} that uses provided {@link SchemaRegistryClient} to connect to the + * schema registry. + * + * @param schemaRegistryClient client to connect to the schema registry */ public ConfluentSchemaRegistryCoder(SchemaRegistryClient schemaRegistryClient) { - this.schemaRegistryClient = schemaRegistryClient; + this(null, schemaRegistryClient, null); + } + + private static boolean parseAutoRegisterSchemas(@Nullable Map registryConfigs) { + Object value = + registryConfigs == null ? null : registryConfigs.get(AUTO_REGISTER_SCHEMAS_CONFIG); + if (value == null) { + return true; + } + if (value instanceof Boolean) { + return (Boolean) value; + } + String stringValue = value.toString(); + if ("true".equalsIgnoreCase(stringValue)) { + return true; + } + if ("false".equalsIgnoreCase(stringValue)) { + return false; + } + throw new IllegalArgumentException( + format( + "Invalid value '%s' for configuration '%s'. Supported values are 'true' or 'false'.", + value, AUTO_REGISTER_SCHEMAS_CONFIG)); } @Override @@ -81,13 +127,32 @@ public Schema readSchema(InputStream in) throws IOException { @Override public void writeSchema(Schema schema, OutputStream out) throws IOException { - try { - int registeredId = schemaRegistryClient.register(subject, schema); - out.write(CONFLUENT_MAGIC_BYTE); - byte[] schemaIdBytes = ByteBuffer.allocate(4).putInt(registeredId).array(); - out.write(schemaIdBytes); - } catch (RestClientException e) { - throw new IOException("Could not register schema in registry", e); + int registeredId; + if (autoRegisterSchemas) { + try { + registeredId = schemaRegistryClient.register(subject, schema); + } catch (RestClientException e) { + throw new IOException( + format( + "Could not register schema %s under subject %s in registry", + schema.getFullName(), subject), + e); + } + } else { + try { + registeredId = schemaRegistryClient.getId(subject, schema); + } catch (RestClientException | IOException e) { + throw new IOException( + format( + "Could not retrieve id for schema %s under subject %s from registry. " + + "Since '%s' is 'false', an identical schema must be registered " + + "under this subject before it can be used", + schema.getFullName(), subject, AUTO_REGISTER_SCHEMAS_CONFIG), + e); + } } + out.write(CONFLUENT_MAGIC_BYTE); + byte[] schemaIdBytes = ByteBuffer.allocate(4).putInt(registeredId).array(); + out.write(schemaIdBytes); } } diff --git a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java index e43d5a76e06817..806ec45ff36990 100644 --- a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java +++ b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java @@ -60,6 +60,7 @@ import java.util.stream.Stream; import static java.lang.String.format; +import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.AUTO_REGISTER_SCHEMAS; import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.BASIC_AUTH_CREDENTIALS_SOURCE; import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.BASIC_AUTH_USER_INFO; import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.BEARER_AUTH_CREDENTIALS_SOURCE; @@ -186,6 +187,7 @@ public Set> optionalOptions() { options.add(BASIC_AUTH_USER_INFO); options.add(BEARER_AUTH_CREDENTIALS_SOURCE); options.add(BEARER_AUTH_TOKEN); + options.add(AUTO_REGISTER_SCHEMAS); return options; } @@ -203,7 +205,8 @@ public Set> forwardOptions() { BASIC_AUTH_CREDENTIALS_SOURCE, BASIC_AUTH_USER_INFO, BEARER_AUTH_CREDENTIALS_SOURCE, - BEARER_AUTH_TOKEN) + BEARER_AUTH_TOKEN, + AUTO_REGISTER_SCHEMAS) .collect(Collectors.toSet()); } @@ -237,6 +240,9 @@ public Set> forwardOptions() { formatOptions .getOptional(BEARER_AUTH_TOKEN) .ifPresent(v -> properties.put("bearer.auth.token", v)); + formatOptions + .getOptional(AUTO_REGISTER_SCHEMAS) + .ifPresent(v -> properties.put("auto.register.schemas", Boolean.toString(v))); if (properties.isEmpty()) { return null; diff --git a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java index 5925f23f763250..aab877a2471e95 100644 --- a/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java +++ b/flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java @@ -51,6 +51,7 @@ import java.util.Set; import static java.lang.String.format; +import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.AUTO_REGISTER_SCHEMAS; import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.BASIC_AUTH_CREDENTIALS_SOURCE; import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.BASIC_AUTH_USER_INFO; import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.BEARER_AUTH_CREDENTIALS_SOURCE; @@ -178,6 +179,7 @@ public Set> optionalOptions() { options.add(BASIC_AUTH_USER_INFO); options.add(BEARER_AUTH_CREDENTIALS_SOURCE); options.add(BEARER_AUTH_TOKEN); + options.add(AUTO_REGISTER_SCHEMAS); return options; } diff --git a/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoderTest.java b/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoderTest.java index 2c93348b002a5a..0fc08a63611903 100644 --- a/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoderTest.java +++ b/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/ConfluentSchemaRegistryCoderTest.java @@ -19,15 +19,20 @@ package org.apache.flink.formats.avro.registry.confluent; import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; import org.apache.avro.Schema; import org.apache.avro.SchemaBuilder; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; +import static java.util.Collections.singletonMap; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -56,6 +61,106 @@ void testSpecificRecordWithConfluentSchemaRegistry() throws Exception { assertThat(byteInStream).isEmpty(); } + @Test + void testWriteSchemaRegistersSchemaByDefault() throws Exception { + MockSchemaRegistryClient client = new MockSchemaRegistryClient(); + Schema schema = + SchemaBuilder.record("testRecord").fields().optionalString("testField").endRecord(); + + ConfluentSchemaRegistryCoder coder = + new ConfluentSchemaRegistryCoder("testSubject", client); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + coder.writeSchema(schema, out); + + assertThat(client.getAllSubjects()).contains("testSubject"); + assertWrittenId(out, client.getId("testSubject", schema)); + } + + @ParameterizedTest + @ValueSource(strings = {"false", "FALSE"}) + void testWriteSchemaWithAutoRegisterDisabledLooksUpExistingId(String configValue) + throws Exception { + MockSchemaRegistryClient client = new MockSchemaRegistryClient(); + Schema schema = + SchemaBuilder.record("testRecord").fields().optionalString("testField").endRecord(); + int schemaId = client.register("testSubject", schema); + + ConfluentSchemaRegistryCoder coder = + new ConfluentSchemaRegistryCoder( + "testSubject", client, singletonMap("auto.register.schemas", configValue)); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + coder.writeSchema(schema, out); + + assertWrittenId(out, schemaId); + } + + @Test + void testWriteSchemaWithAutoRegisterDisabledDoesNotRegisterUnknownSchema() throws Exception { + MockSchemaRegistryClient client = new MockSchemaRegistryClient(); + Schema schema = + SchemaBuilder.record("testRecord").fields().optionalString("testField").endRecord(); + + ConfluentSchemaRegistryCoder coder = + new ConfluentSchemaRegistryCoder( + "unknownSubject", client, singletonMap("auto.register.schemas", "false")); + + assertThatThrownBy(() -> coder.writeSchema(schema, new ByteArrayOutputStream())) + .isInstanceOf(IOException.class) + .hasMessageContaining("unknownSubject"); + // the failed lookup must not have registered anything under the subject + assertThatThrownBy(() -> client.getAllVersions("unknownSubject")) + .isInstanceOf(RestClientException.class) + .hasMessageContaining("Subject Not Found"); + } + + @Test + void testWriteSchemaWithAutoRegisterDisabledViaBooleanConfig() throws Exception { + // DataStream API users can pass Boolean values instead of Strings in the registry configs + MockSchemaRegistryClient client = new MockSchemaRegistryClient(); + Schema schema = + SchemaBuilder.record("testRecord").fields().optionalString("testField").endRecord(); + int schemaId = client.register("testSubject", schema); + + ConfluentSchemaRegistryCoder coder = + new ConfluentSchemaRegistryCoder( + "testSubject", + client, + singletonMap("auto.register.schemas", Boolean.FALSE)); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + coder.writeSchema(schema, out); + + assertWrittenId(out, schemaId); + } + + @Test + void testWriteSchemaWithAutoRegisterEnabledViaBooleanConfig() throws Exception { + MockSchemaRegistryClient client = new MockSchemaRegistryClient(); + Schema schema = + SchemaBuilder.record("testRecord").fields().optionalString("testField").endRecord(); + + ConfluentSchemaRegistryCoder coder = + new ConfluentSchemaRegistryCoder( + "testSubject", client, singletonMap("auto.register.schemas", Boolean.TRUE)); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + coder.writeSchema(schema, out); + + assertThat(client.getAllSubjects()).contains("testSubject"); + } + + @Test + void testInvalidAutoRegisterSchemasValueIsRejected() { + MockSchemaRegistryClient client = new MockSchemaRegistryClient(); + + assertThatThrownBy( + () -> + new ConfluentSchemaRegistryCoder( + "testSubject", + client, + singletonMap("auto.register.schemas", "enabled"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("auto.register.schemas"); + } + @Test void testMagicByteVerification() throws Exception { MockSchemaRegistryClient client = new MockSchemaRegistryClient(); @@ -74,4 +179,11 @@ void testMagicByteVerification() throws Exception { .isInstanceOf(IOException.class); } } + + private static void assertWrittenId(ByteArrayOutputStream out, int expectedId) + throws IOException { + DataInputStream in = new DataInputStream(new ByteArrayInputStream(out.toByteArray())); + assertThat(in.readByte()).isEqualTo((byte) 0); + assertThat(in.readInt()).isEqualTo(expectedId); + } } diff --git a/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactoryTest.java b/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactoryTest.java index 6ace72b199ed87..08290c18b0ccf9 100644 --- a/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactoryTest.java +++ b/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactoryTest.java @@ -39,6 +39,8 @@ import org.apache.avro.Schema.Parser; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.util.HashMap; import java.util.Map; @@ -229,6 +231,92 @@ public void testSerializationSchemaWithInvalidOptionalSchema() { null, SCHEMA.toPhysicalRowDataType())); } + @ParameterizedTest + @ValueSource(strings = {"true", "false"}) + void testSeDeSchemaWithAutoRegisterSchemas(String autoRegisterSchemas) { + final Map expectedProperties = new HashMap<>(); + expectedProperties.put("auto.register.schemas", autoRegisterSchemas); + + final Map options = + getModifiedOptions( + opts -> + opts.put( + "avro-confluent.auto.register.schemas", + autoRegisterSchemas)); + + final AvroRowDataDeserializationSchema expectedDeser = + new AvroRowDataDeserializationSchema( + ConfluentRegistryAvroDeserializationSchema.forGeneric( + AvroSchemaConverter.convertToSchema(ROW_TYPE), + REGISTRY_URL, + expectedProperties), + AvroToRowDataConverters.createRowConverter(ROW_TYPE), + InternalTypeInfo.of(ROW_TYPE)); + + final DynamicTableSource actualSource = createTableSource(SCHEMA, options); + assertThat(actualSource).isInstanceOf(TestDynamicTableFactory.DynamicTableSourceMock.class); + TestDynamicTableFactory.DynamicTableSourceMock scanSourceMock = + (TestDynamicTableFactory.DynamicTableSourceMock) actualSource; + + DeserializationSchema actualDeser = + scanSourceMock.valueFormat.createRuntimeDecoder( + ScanRuntimeProviderContext.INSTANCE, SCHEMA.toPhysicalRowDataType()); + assertThat(actualDeser).isEqualTo(expectedDeser); + + final AvroRowDataSerializationSchema expectedSer = + new AvroRowDataSerializationSchema( + ROW_TYPE, + ConfluentRegistryAvroSerializationSchema.forGeneric( + SUBJECT, + AvroSchemaConverter.convertToSchema(ROW_TYPE), + REGISTRY_URL, + expectedProperties), + RowDataToAvroConverters.createConverter(ROW_TYPE)); + + final DynamicTableSink actualSink = createTableSink(SCHEMA, options); + assertThat(actualSink).isInstanceOf(TestDynamicTableFactory.DynamicTableSinkMock.class); + TestDynamicTableFactory.DynamicTableSinkMock sinkMock = + (TestDynamicTableFactory.DynamicTableSinkMock) actualSink; + + SerializationSchema actualSer = + sinkMock.valueFormat.createRuntimeEncoder(null, SCHEMA.toPhysicalRowDataType()); + assertThat(actualSer).isEqualTo(expectedSer); + } + + @Test + void testAutoRegisterSchemasViaPropertiesMap() { + // auto.register.schemas can also be provided through the generic properties map, + // without the dedicated table option + final Map expectedProperties = new HashMap<>(); + expectedProperties.put("auto.register.schemas", "false"); + + final Map options = + getModifiedOptions( + opts -> + opts.put( + "avro-confluent.properties.auto.register.schemas", + "false")); + + final AvroRowDataSerializationSchema expectedSer = + new AvroRowDataSerializationSchema( + ROW_TYPE, + ConfluentRegistryAvroSerializationSchema.forGeneric( + SUBJECT, + AvroSchemaConverter.convertToSchema(ROW_TYPE), + REGISTRY_URL, + expectedProperties), + RowDataToAvroConverters.createConverter(ROW_TYPE)); + + final DynamicTableSink actualSink = createTableSink(SCHEMA, options); + assertThat(actualSink).isInstanceOf(TestDynamicTableFactory.DynamicTableSinkMock.class); + TestDynamicTableFactory.DynamicTableSinkMock sinkMock = + (TestDynamicTableFactory.DynamicTableSinkMock) actualSink; + + SerializationSchema actualSer = + sinkMock.valueFormat.createRuntimeEncoder(null, SCHEMA.toPhysicalRowDataType()); + assertThat(actualSer).isEqualTo(expectedSer); + } + // ------------------------------------------------------------------------ // Utilities // ------------------------------------------------------------------------ diff --git a/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactoryTest.java b/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactoryTest.java index e7a263ab42123b..52f1d942d00e1f 100644 --- a/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactoryTest.java +++ b/flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactoryTest.java @@ -168,6 +168,31 @@ public void testSeDeSchemaWithSchemaOption() { assertThat(actualSer).isEqualTo(expectedSer); } + @Test + void testSeDeSchemaWithAutoRegisterSchemas() { + final Map options = getAllOptions(); + options.put("debezium-avro-confluent.auto.register.schemas", "false"); + + final Map registryConfigs = getRegistryConfigs(); + registryConfigs.put("auto.register.schemas", "false"); + + DebeziumAvroDeserializationSchema expectedDeser = + new DebeziumAvroDeserializationSchema( + ROW_TYPE, + InternalTypeInfo.of(ROW_TYPE), + REGISTRY_URL, + null, + registryConfigs); + DeserializationSchema actualDeser = createDeserializationSchema(options); + assertThat(actualDeser).isEqualTo(expectedDeser); + + DebeziumAvroSerializationSchema expectedSer = + new DebeziumAvroSerializationSchema( + ROW_TYPE, REGISTRY_URL, SUBJECT, null, registryConfigs); + SerializationSchema actualSer = createSerializationSchema(options); + assertThat(actualSer).isEqualTo(expectedSer); + } + @Test public void testSeDeSchemaWithInvalidSchemaOption() { final Map options = getAllOptions(); diff --git a/flink-formats/flink-sql-avro-confluent-registry/src/main/resources/META-INF/NOTICE b/flink-formats/flink-sql-avro-confluent-registry/src/main/resources/META-INF/NOTICE index b57a307e30c04f..c8ba6f6c2063c3 100644 --- a/flink-formats/flink-sql-avro-confluent-registry/src/main/resources/META-INF/NOTICE +++ b/flink-formats/flink-sql-avro-confluent-registry/src/main/resources/META-INF/NOTICE @@ -9,18 +9,19 @@ This project bundles the following dependencies under the Apache Software Licens - com.fasterxml.jackson.core:jackson-annotations:2.22 - com.fasterxml.jackson.core:jackson-core:2.22.2 - com.fasterxml.jackson.core:jackson-databind:2.22.2 +- com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.22.2 - com.google.guava:guava:32.0.1-jre - commons-io:commons-io:2.15.1 -- io.confluent:common-utils:7.5.3 -- io.confluent:kafka-schema-registry-client:7.5.3 +- io.confluent:common-utils:7.9.8 +- io.confluent:kafka-schema-registry-client:7.9.8 - org.apache.avro:avro:1.11.5 - org.apache.commons:commons-compress:1.26.0 - org.apache.commons:commons-lang3:3.18.0 -- org.apache.kafka:kafka-clients:7.5.3-ccs +- org.apache.kafka:kafka-clients:7.9.8-ccs - org.xerial.snappy:snappy-java:1.1.10.7 - org.yaml:snakeyaml:2.3 This project bundles the following dependencies under the BSD license. See bundled license files for details. -- com.github.luben:zstd-jni:1.5.5-1 +- com.github.luben:zstd-jni:1.5.6-4 diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java index 655f4e3b4666ab..fa8a82913b135b 100644 --- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java +++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/util/DockerImageVersions.java @@ -28,9 +28,9 @@ */ public class DockerImageVersions { - public static final String KAFKA = "confluentinc/cp-kafka:7.5.3"; + public static final String KAFKA = "confluentinc/cp-kafka:8.3.0"; - public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:7.5.3"; + public static final String SCHEMA_REGISTRY = "confluentinc/cp-schema-registry:8.3.0"; public static final String KINESALITE = "instructure/kinesalite:latest";