[FLINK-23450][avro-confluent-registry] Set properties map for DebeziumAvroFormat

Author: twalthr

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

@@ -42,6 +42,8 @@
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
 
+import javax.annotation.Nullable;
+
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -84,14 +86,10 @@ public DeserializationSchema<RowData> createRuntimeDecoder(
                 final TypeInformation<RowData> rowDataTypeInfo =
                         context.createTypeInformation(producedDataType);
                 return new AvroRowDataDeserializationSchema(
-                        optionalPropertiesMap.isEmpty()
-                                ? ConfluentRegistryAvroDeserializationSchema.forGeneric(
-                                        AvroSchemaConverter.convertToSchema(rowType),
-                                        schemaRegistryURL)
-                                : ConfluentRegistryAvroDeserializationSchema.forGeneric(
-                                        AvroSchemaConverter.convertToSchema(rowType),
-                                        schemaRegistryURL,
-                                        optionalPropertiesMap),
+                        ConfluentRegistryAvroDeserializationSchema.forGeneric(
+                                AvroSchemaConverter.convertToSchema(rowType),
+                                schemaRegistryURL,
+                                optionalPropertiesMap),
                         AvroToRowDataConverters.createRowConverter(rowType),
                         rowDataTypeInfo);
             }
@@ -126,16 +124,11 @@ public SerializationSchema<RowData> createRuntimeEncoder(
                 final RowType rowType = (RowType) consumedDataType.getLogicalType();
                 return new AvroRowDataSerializationSchema(
                         rowType,
-                        optionalPropertiesMap.isEmpty()
-                                ? ConfluentRegistryAvroSerializationSchema.forGeneric(
-                                        subject.get(),
-                                        AvroSchemaConverter.convertToSchema(rowType),
-                                        schemaRegistryURL)
-                                : ConfluentRegistryAvroSerializationSchema.forGeneric(
-                                        subject.get(),
-                                        AvroSchemaConverter.convertToSchema(rowType),
-                                        schemaRegistryURL,
-                                        optionalPropertiesMap),
+                        ConfluentRegistryAvroSerializationSchema.forGeneric(
+                                subject.get(),
+                                AvroSchemaConverter.convertToSchema(rowType),
+                                schemaRegistryURL,
+                                optionalPropertiesMap),
                         RowDataToAvroConverters.createConverter(rowType));
             }
 
@@ -173,7 +166,8 @@ public Set<ConfigOption<?>> optionalOptions() {
         return options;
     }
 
-    private Map<String, String> buildOptionalPropertiesMap(ReadableConfig formatOptions) {
+    public static @Nullable Map<String, String> buildOptionalPropertiesMap(
+            ReadableConfig formatOptions) {
         final Map<String, String> properties = new HashMap<>();
 
         formatOptions
@@ -201,6 +195,9 @@ private Map<String, String> buildOptionalPropertiesMap(ReadableConfig formatOpti
                 .getOptional(BEARER_AUTH_TOKEN)
                 .ifPresent(v -> properties.put("bearer.auth.token", v));
 
+        if (properties.isEmpty()) {
+            return null;
+        }
         return properties;
     }
 }

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

@@ -34,7 +34,10 @@
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Collector;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
+import java.util.Map;
 import java.util.Objects;
 
 import static java.lang.String.format;
@@ -75,14 +78,23 @@ public final class DebeziumAvroDeserializationSchema implements DeserializationS
 
     public DebeziumAvroDeserializationSchema(
             RowType rowType, TypeInformation<RowData> producedTypeInfo, String schemaRegistryUrl) {
+        this(rowType, producedTypeInfo, schemaRegistryUrl, null);
+    }
+
+    public DebeziumAvroDeserializationSchema(
+            RowType rowType,
+            TypeInformation<RowData> producedTypeInfo,
+            String schemaRegistryUrl,
+            @Nullable Map<String, ?> registryConfigs) {
         this.producedTypeInfo = producedTypeInfo;
         RowType debeziumAvroRowType = createDebeziumAvroRowType(fromLogicalToDataType(rowType));
 
         this.avroDeserializer =
                 new AvroRowDataDeserializationSchema(
                         ConfluentRegistryAvroDeserializationSchema.forGeneric(
                                 AvroSchemaConverter.convertToSchema(debeziumAvroRowType),
-                                schemaRegistryUrl),
+                                schemaRegistryUrl,
+                                registryConfigs),
                         AvroToRowDataConverters.createRowConverter(debeziumAvroRowType),
                         producedTypeInfo);
     }

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

@@ -39,11 +39,21 @@
 import org.apache.flink.types.RowKind;
 
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroFormatFactory.buildOptionalPropertiesMap;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.BASIC_AUTH_CREDENTIALS_SOURCE;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.BASIC_AUTH_USER_INFO;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.BEARER_AUTH_CREDENTIALS_SOURCE;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.BEARER_AUTH_TOKEN;
 import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.SCHEMA_REGISTRY_SUBJECT;
 import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.SCHEMA_REGISTRY_URL;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.SSL_KEYSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.SSL_KEYSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroOptions.SSL_TRUSTSTORE_PASSWORD;
 
 /**
  * Format factory for providing configured instances of Debezium Avro to RowData {@link
@@ -59,7 +69,9 @@ public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
             DynamicTableFactory.Context context, ReadableConfig formatOptions) {
 
         FactoryUtil.validateFactoryOptions(this, formatOptions);
+
         String schemaRegistryURL = formatOptions.get(SCHEMA_REGISTRY_URL);
+        Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
 
         return new DecodingFormat<DeserializationSchema<RowData>>() {
             @Override
@@ -69,7 +81,7 @@ public DeserializationSchema<RowData> createRuntimeDecoder(
                 final TypeInformation<RowData> producedTypeInfo =
                         context.createTypeInformation(producedDataType);
                 return new DebeziumAvroDeserializationSchema(
-                        rowType, producedTypeInfo, schemaRegistryURL);
+                        rowType, producedTypeInfo, schemaRegistryURL, optionalPropertiesMap);
             }
 
             @Override
@@ -89,8 +101,11 @@ public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
             DynamicTableFactory.Context context, ReadableConfig formatOptions) {
 
         FactoryUtil.validateFactoryOptions(this, formatOptions);
+
         String schemaRegistryURL = formatOptions.get(SCHEMA_REGISTRY_URL);
         Optional<String> subject = formatOptions.getOptional(SCHEMA_REGISTRY_SUBJECT);
+        Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+
         if (!subject.isPresent()) {
             throw new ValidationException(
                     String.format(
@@ -114,7 +129,7 @@ public SerializationSchema<RowData> createRuntimeEncoder(
                     DynamicTableSink.Context context, DataType consumedDataType) {
                 final RowType rowType = (RowType) consumedDataType.getLogicalType();
                 return new DebeziumAvroSerializationSchema(
-                        rowType, schemaRegistryURL, subject.get());
+                        rowType, schemaRegistryURL, subject.get(), optionalPropertiesMap);
             }
         };
     }
@@ -135,6 +150,14 @@ public Set<ConfigOption<?>> requiredOptions() {
     public Set<ConfigOption<?>> optionalOptions() {
         Set<ConfigOption<?>> options = new HashSet<>();
         options.add(SCHEMA_REGISTRY_SUBJECT);
+        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/java/org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroSerializationSchema.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.formats.avro.registry.confluent.debezium;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
@@ -31,6 +32,9 @@
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
 
+import javax.annotation.Nullable;
+
+import java.util.Map;
 import java.util.Objects;
 
 import static java.lang.String.format;
@@ -40,6 +44,7 @@
  * Serialization schema from Flink Table/SQL internal data structure {@link RowData} to Debezium
  * Avro.
  */
+@Internal
 public class DebeziumAvroSerializationSchema implements SerializationSchema<RowData> {
     private static final long serialVersionUID = 1L;
 
@@ -55,6 +60,14 @@ public class DebeziumAvroSerializationSchema implements SerializationSchema<RowD
 
     public DebeziumAvroSerializationSchema(
             RowType rowType, String schemaRegistryUrl, String schemaRegistrySubject) {
+        this(rowType, schemaRegistryUrl, schemaRegistrySubject, null);
+    }
+
+    public DebeziumAvroSerializationSchema(
+            RowType rowType,
+            String schemaRegistryUrl,
+            String schemaRegistrySubject,
+            @Nullable Map<String, ?> registryConfigs) {
         RowType debeziumAvroRowType = createDebeziumAvroRowType(fromLogicalToDataType(rowType));
 
         this.avroSerializer =
@@ -63,7 +76,8 @@ public DebeziumAvroSerializationSchema(
                         ConfluentRegistryAvroSerializationSchema.forGeneric(
                                 schemaRegistrySubject,
                                 AvroSchemaConverter.convertToSchema(debeziumAvroRowType),
-                                schemaRegistryUrl),
+                                schemaRegistryUrl,
+                                registryConfigs),
                         RowDataToAvroConverters.createConverter(debeziumAvroRowType));
     }
 

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

@@ -68,14 +68,19 @@ public class DebeziumAvroFormatFactoryTest extends TestLogger {
     public void testSeDeSchema() {
         final Map<String, String> options = getAllOptions();
 
+        final Map<String, String> registryConfigs = new HashMap<>();
+        registryConfigs.put("basic.auth.user.info", "something1");
+        registryConfigs.put("basic.auth.credentials.source", "something2");
+
         DebeziumAvroDeserializationSchema expectedDeser =
                 new DebeziumAvroDeserializationSchema(
-                        ROW_TYPE, InternalTypeInfo.of(ROW_TYPE), REGISTRY_URL);
+                        ROW_TYPE, InternalTypeInfo.of(ROW_TYPE), REGISTRY_URL, registryConfigs);
         DeserializationSchema<RowData> actualDeser = createDeserializationSchema(options);
         assertEquals(expectedDeser, actualDeser);
 
         DebeziumAvroSerializationSchema expectedSer =
-                new DebeziumAvroSerializationSchema(ROW_TYPE, REGISTRY_URL, SUBJECT);
+                new DebeziumAvroSerializationSchema(
+                        ROW_TYPE, REGISTRY_URL, SUBJECT, registryConfigs);
         SerializationSchema<RowData> actualSer = createSerializationSchema(options);
         Assert.assertEquals(expectedSer, actualSer);
     }
@@ -89,6 +94,8 @@ private Map<String, String> getAllOptions() {
         options.put("format", DebeziumAvroFormatFactory.IDENTIFIER);
         options.put("debezium-avro-confluent.schema-registry.url", REGISTRY_URL);
         options.put("debezium-avro-confluent.schema-registry.subject", SUBJECT);
+        options.put("debezium-avro-confluent.basic-auth.user-info", "something1");
+        options.put("debezium-avro-confluent.basic-auth.credentials-source", "something2");
         return options;
     }