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

…mAvroFormat This closes apache#16565.
static-max · Aug 2, 2021 · 78db0e7 · 78db0e7
1 parent 4b517a9
commit 78db0e7
Show file tree

Hide file tree

Showing 6 changed files with 71 additions and 28 deletions.
diff --git a/.../java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java b/.../java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactoryTest.java
@@ -621,7 +621,7 @@ private SerializationSchema<RowData> createConfluentAvroSerSchema(
 
  private SerializationSchema<RowData> createDebeziumAvroSerSchema(
  RowType rowType, String subject) {
- return new DebeziumAvroSerializationSchema(rowType, TEST_REGISTRY_URL, subject);
+ return new DebeziumAvroSerializationSchema(rowType, TEST_REGISTRY_URL, subject, null);
  }
 
  // --------------------------------------------------------------------------------------------

diff --git a/...main/java/org/apache/flink/formats/avro/registry/confluent/RegistryAvroFormatFactory.java b/...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;
@@ -85,14 +87,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);
  }
@@ -127,16 +125,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));
  }
 
@@ -175,7 +168,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.getOptional(PROPERTIES).ifPresent(properties::putAll);
@@ -205,6 +199,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;
  }
 }
diff --git a/...che/flink/formats/avro/registry/confluent/debezium/DebeziumAvroDeserializationSchema.java b/...che/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;
@@ -74,15 +77,19 @@ public final class DebeziumAvroDeserializationSchema implements DeserializationS
  private final TypeInformation<RowData> producedTypeInfo;
 
  public DebeziumAvroDeserializationSchema(
- RowType rowType, TypeInformation<RowData> producedTypeInfo, String schemaRegistryUrl) {
+ 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);
  }

diff --git a/.../org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java b/.../org/apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactory.java
@@ -39,11 +39,22 @@
 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.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.SUBJECT;
 import static org.apache.flink.formats.avro.registry.confluent.AvroConfluentFormatOptions.URL;
+import static org.apache.flink.formats.avro.registry.confluent.RegistryAvroFormatFactory.buildOptionalPropertiesMap;
 
 /**
  * Format factory for providing configured instances of Debezium Avro to RowData {@link
@@ -60,6 +71,7 @@ public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
 
  FactoryUtil.validateFactoryOptions(this, formatOptions);
  String schemaRegistryURL = formatOptions.get(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
@@ -91,6 +103,8 @@ public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
  FactoryUtil.validateFactoryOptions(this, formatOptions);
  String schemaRegistryURL = formatOptions.get(URL);
  Optional<String> subject = formatOptions.getOptional(SUBJECT);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+
  if (!subject.isPresent()) {
  throw new ValidationException(
  String.format(
@@ -114,7 +128,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 +149,15 @@ public Set<ConfigOption<?>> requiredOptions() {
  public Set<ConfigOption<?>> optionalOptions() {
  Set<ConfigOption<?>> options = new HashSet<>();
  options.add(SUBJECT);
+ 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;
  }
 }
diff --git a/...pache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroSerializationSchema.java b/...pache/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;
 
@@ -54,7 +59,10 @@ public class DebeziumAvroSerializationSchema implements SerializationSchema<RowD
  private transient GenericRowData outputReuse;
 
  public DebeziumAvroSerializationSchema(
- RowType rowType, String schemaRegistryUrl, String schemaRegistrySubject) {
+ RowType rowType,
+ String schemaRegistryUrl,
+ String schemaRegistrySubject,
+ @Nullable Map<String, ?> registryConfigs) {
  RowType debeziumAvroRowType = createDebeziumAvroRowType(fromLogicalToDataType(rowType));
 
  this.avroSerializer =
@@ -63,7 +71,8 @@ public DebeziumAvroSerializationSchema(
  ConfluentRegistryAvroSerializationSchema.forGeneric(
  schemaRegistrySubject,
  AvroSchemaConverter.convertToSchema(debeziumAvroRowType),
- schemaRegistryUrl),
+ schemaRegistryUrl,
+ registryConfigs),
  RowDataToAvroConverters.createConverter(debeziumAvroRowType));
  }
 

diff --git a/.../apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactoryTest.java b/.../apache/flink/formats/avro/registry/confluent/debezium/DebeziumAvroFormatFactoryTest.java
@@ -67,14 +67,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);
  }
@@ -88,6 +93,8 @@ private Map<String, String> getAllOptions() {
  options.put("format", DebeziumAvroFormatFactory.IDENTIFIER);
  options.put("debezium-avro-confluent.url", REGISTRY_URL);
  options.put("debezium-avro-confluent.subject", SUBJECT);
+ options.put("debezium-avro-confluent.basic-auth.user-info", "something1");
+ options.put("debezium-avro-confluent.basic-auth.credentials-source", "something2");
  return options;
  }