[FLINK-39605][Flink-Formats] Add avro variant confluent format

flink-formats/flink-avro-confluent-registry/src/main/java/org/apache/flink/formats/avro/registry/confluent/ConfluentRegistryAvroVariantFormatFactory.java

@@ -0,0 +1,115 @@
+/*
+ * 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.formats.avro.registry.confluent;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.formats.avro.AvroVariantDecodingFormat;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+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;
+import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.BEARER_AUTH_TOKEN;
+import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.SSL_KEYSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.SSL_KEYSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.SSL_TRUSTSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.URL;
+
+/**
+ * Table format factory for providing configured instances of Confluent Schema Registry Avro to
+ * Variant {@link DeserializationSchema}. Deserialization only — no serialization support.
+ */
+@Internal
+public class ConfluentRegistryAvroVariantFormatFactory implements DeserializationFormatFactory {
+
+    public static final String IDENTIFIER = "avro-variant-confluent";
+
+    private static final int SCHEMA_CACHE_CAPACITY = 1000;
+
+    @Override
+    public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
+            DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+        FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+        String schemaRegistryURL = formatOptions.get(URL);
+        Map<String, String> optionalPropertiesMap =
+                RegistryAvroFormatFactory.buildOptionalPropertiesMap(formatOptions);
+
+        return new AvroVariantDecodingFormat(
+                new CachedSchemaCoderProvider(
+                        null, schemaRegistryURL, SCHEMA_CACHE_CAPACITY, optionalPropertiesMap),
+                SCHEMA_CACHE_CAPACITY);
+    }
+
+    @Override
+    public String factoryIdentifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public Set<ConfigOption<?>> requiredOptions() {
+        Set<ConfigOption<?>> options = new HashSet<>();
+        options.add(URL);
+        return options;
+    }
+
+    @Override
+    public Set<ConfigOption<?>> optionalOptions() {
+        Set<ConfigOption<?>> options = new HashSet<>();
+        options.add(PROPERTIES);
+        options.add(SSL_KEYSTORE_LOCATION);
+        options.add(SSL_KEYSTORE_PASSWORD);
+        options.add(SSL_TRUSTSTORE_LOCATION);
+        options.add(SSL_TRUSTSTORE_PASSWORD);
+        options.add(BASIC_AUTH_CREDENTIALS_SOURCE);
+        options.add(BASIC_AUTH_USER_INFO);
+        options.add(BEARER_AUTH_CREDENTIALS_SOURCE);
+        options.add(BEARER_AUTH_TOKEN);
+        return options;
+    }
+
+    @Override
+    public Set<ConfigOption<?>> forwardOptions() {
+        Set<ConfigOption<?>> options = new HashSet<>();
+        options.add(URL);
+        options.add(PROPERTIES);
+        options.add(SSL_KEYSTORE_LOCATION);
+        options.add(SSL_KEYSTORE_PASSWORD);
+        options.add(SSL_TRUSTSTORE_LOCATION);
+        options.add(SSL_TRUSTSTORE_PASSWORD);
+        options.add(BASIC_AUTH_CREDENTIALS_SOURCE);
+        options.add(BASIC_AUTH_USER_INFO);
+        options.add(BEARER_AUTH_CREDENTIALS_SOURCE);
+        options.add(BEARER_AUTH_TOKEN);
+        return options;
+    }
+}

flink-formats/flink-avro-confluent-registry/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory

@@ -15,3 +15,4 @@
 
 org.apache.flink.formats.avro.registry.confluent.RegistryAvroFormatFactory
 org.apache.flink.formats.avro.registry.confluent.debezium.DebeziumAvroFormatFactory
+org.apache.flink.formats.avro.registry.confluent.ConfluentRegistryAvroVariantFormatFactory

flink-formats/flink-avro-confluent-registry/src/test/java/org/apache/flink/formats/avro/registry/confluent/ConfluentRegistryAvroVariantFormatFactoryTest.java

@@ -0,0 +1,177 @@
+/*
+ * 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.formats.avro.registry.confluent;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.formats.avro.AvroVariantDeserializationSchema;
+import org.apache.flink.formats.avro.RegistryWriterAvroDeserializationSchema;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.TestDynamicTableFactory;
+import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.VariantType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSource;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ConfluentRegistryAvroVariantFormatFactoryTest {
+
+    private static final ResolvedSchema SCHEMA =
+            ResolvedSchema.of(Column.physical("data", DataTypes.VARIANT()));
+
+    private static final String REGISTRY_URL = "http://localhost:8081";
+    private static final int SCHEMA_CACHE_CAPACITY = 1000;
+
+    @Test
+    void testDecodingFormat() {
+        final AvroVariantDeserializationSchema expectedDeser =
+                new AvroVariantDeserializationSchema(
+                        new RegistryWriterAvroDeserializationSchema(
+                                new CachedSchemaCoderProvider(
+                                        null, REGISTRY_URL, SCHEMA_CACHE_CAPACITY, null)),
+                        false,
+                        SCHEMA_CACHE_CAPACITY,
+                        InternalTypeInfo.of(RowType.of(new VariantType())));
+
+        final DynamicTableSource actualSource = createTableSource(SCHEMA, getDefaultOptions());
+        assertThat(actualSource).isInstanceOf(TestDynamicTableFactory.DynamicTableSourceMock.class);
+        TestDynamicTableFactory.DynamicTableSourceMock scanSourceMock =
+                (TestDynamicTableFactory.DynamicTableSourceMock) actualSource;
+
+        DeserializationSchema<RowData> actualDeser =
+                scanSourceMock.valueFormat.createRuntimeDecoder(
+                        ScanRuntimeProviderContext.INSTANCE, SCHEMA.toPhysicalRowDataType());
+
+        assertThat(actualDeser).isEqualTo(expectedDeser);
+    }
+
+    @Test
+    void testDecodingFormatWithOptionalProperties() {
+        Map<String, String> expectedProperties = getExpectedProperties();
+
+        final AvroVariantDeserializationSchema expectedDeser =
+                new AvroVariantDeserializationSchema(
+                        new RegistryWriterAvroDeserializationSchema(
+                                new CachedSchemaCoderProvider(
+                                        null,
+                                        REGISTRY_URL,
+                                        SCHEMA_CACHE_CAPACITY,
+                                        expectedProperties)),
+                        false,
+                        SCHEMA_CACHE_CAPACITY,
+                        InternalTypeInfo.of(RowType.of(new VariantType())));
+
+        final DynamicTableSource actualSource =
+                createTableSource(SCHEMA, getOptionsWithProperties());
+        TestDynamicTableFactory.DynamicTableSourceMock scanSourceMock =
+                (TestDynamicTableFactory.DynamicTableSourceMock) actualSource;
+
+        DeserializationSchema<RowData> actualDeser =
+                scanSourceMock.valueFormat.createRuntimeDecoder(
+                        ScanRuntimeProviderContext.INSTANCE, SCHEMA.toPhysicalRowDataType());
+
+        assertThat(actualDeser).isEqualTo(expectedDeser);
+    }
+
+    @Test
+    void testMissingUrl() {
+        final Map<String, String> options = getDefaultOptions();
+        options.remove("avro-variant-confluent.url");
+
+        assertThatThrownBy(() -> createTableSource(SCHEMA, options))
+                .isInstanceOf(ValidationException.class);
+    }
+
+    @Test
+    void testDecodingFormatWithSchemaMetadata() {
+        final AvroVariantDeserializationSchema expectedDeser =
+                new AvroVariantDeserializationSchema(
+                        new RegistryWriterAvroDeserializationSchema(
+                                new CachedSchemaCoderProvider(
+                                        null, REGISTRY_URL, SCHEMA_CACHE_CAPACITY, null)),
+                        true,
+                        SCHEMA_CACHE_CAPACITY,
+                        InternalTypeInfo.of(
+                                RowType.of(
+                                        new VariantType(),
+                                        new VarCharType(true, VarCharType.MAX_LENGTH))));
+
+        final DynamicTableSource actualSource = createTableSource(SCHEMA, getDefaultOptions());
+        TestDynamicTableFactory.DynamicTableSourceMock scanSourceMock =
+                (TestDynamicTableFactory.DynamicTableSourceMock) actualSource;
+
+        scanSourceMock.valueFormat.applyReadableMetadata(Collections.singletonList("schema"));
+
+        DeserializationSchema<RowData> actualDeser =
+                scanSourceMock.valueFormat.createRuntimeDecoder(
+                        ScanRuntimeProviderContext.INSTANCE, SCHEMA.toPhysicalRowDataType());
+
+        assertThat(actualDeser).isEqualTo(expectedDeser);
+    }
+
+    private Map<String, String> getDefaultOptions() {
+        final Map<String, String> options = new HashMap<>();
+        options.put("connector", TestDynamicTableFactory.IDENTIFIER);
+        options.put("target", "MyTarget");
+        options.put("buffer-size", "1000");
+
+        options.put("format", ConfluentRegistryAvroVariantFormatFactory.IDENTIFIER);
+        options.put("avro-variant-confluent.url", REGISTRY_URL);
+        return options;
+    }
+
+    private Map<String, String> getOptionsWithProperties() {
+        final Map<String, String> options = getDefaultOptions();
+        String prefix = "avro-variant-confluent.";
+        options.put(prefix + "ssl.keystore.location", "/test-keystore.jks");
+        options.put(prefix + "ssl.keystore.password", "123456");
+        options.put(prefix + "ssl.truststore.location", "/test-keystore.jks");
+        options.put(prefix + "ssl.truststore.password", "123456");
+        options.put(prefix + "basic-auth.credentials-source", "USER_INFO");
+        options.put(prefix + "basic-auth.user-info", "user:pwd");
+        options.put(prefix + "properties.bearer.auth.token", "CUSTOM");
+        return options;
+    }
+
+    private Map<String, String> getExpectedProperties() {
+        Map<String, String> expectedProperties = new HashMap<>();
+        expectedProperties.put("schema.registry.ssl.keystore.location", "/test-keystore.jks");
+        expectedProperties.put("schema.registry.ssl.keystore.password", "123456");
+        expectedProperties.put("schema.registry.ssl.truststore.location", "/test-keystore.jks");
+        expectedProperties.put("schema.registry.ssl.truststore.password", "123456");
+        expectedProperties.put("basic.auth.credentials.source", "USER_INFO");
+        expectedProperties.put("basic.auth.user.info", "user:pwd");
+        expectedProperties.put("bearer.auth.token", "CUSTOM");
+        return expectedProperties;
+    }
+}