You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "davidradl (via GitHub)" <gi...@apache.org> on 2024/04/24 16:08:37 UTC
[PR] Draft pr for initial review of apicurio [flink]
davidradl opened a new pull request, #24715:
URL: https://github.com/apache/flink/pull/24715
Early code for review of Apicurio Avor prototyping
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft pr for initial review of apicurio [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on PR #24715:
URL: https://github.com/apache/flink/pull/24715#issuecomment-2117817612
@mnuttall thank you very much for the review I will have a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FLINK-35311 Draft pr for initial review of apicurio [flink]
Posted by "nictownsend (via GitHub)" <gi...@apache.org>.
nictownsend commented on code in PR #24715:
URL: https://github.com/apache/flink/pull/24715#discussion_r1618968305
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioRegistryAvroFormatFactory.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.formats.avro.AvroRowDataDeserializationSchema;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
+import org.apache.flink.formats.avro.AvroToRowDataConverters;
+import org.apache.flink.formats.avro.RowDataToAvroConverters;
+import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.Projection;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.format.ProjectableDecodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.GenericRowData;
+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 org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.avro.Schema;
+import org.apache.avro.Schema.Parser;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.lang.String.format;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_SECRET;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_SCOPE;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_DESCRIPTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_NAME;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_VERSION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SCHEMA;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Apicurio avro format. */
+@Internal
+public class ApicurioRegistryAvroFormatFactory
+ implements DeserializationFormatFactory, SerializationFormatFactory {
+
+ // used this key rather than apicurio-avro (the format name) to be consistent with Confluent.
+ public static final String IDENTIFIER = "avro-apicurio";
+
+ @Override
+ public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+ String schemaRegistryURL = formatOptions.get(URL);
+ Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+ return new ProjectableDecodingFormat<DeserializationSchema<RowData>>() {
+ // --------------------------------------------------------------------------------------------
+ // Mutable attributes
+ // --------------------------------------------------------------------------------------------
+
+ private List<String> metadataKeys;
+
+ @Override
+ public DeserializationSchema<RowData> createRuntimeDecoder(
+ DynamicTableSource.Context context,
+ DataType producedDataType,
+ int[][] projections) {
+ producedDataType = Projection.of(projections).project(producedDataType);
+ final RowType rowType = (RowType) producedDataType.getLogicalType();
+ // schema has been supplied
+ final Schema schema =
+ schemaString
+ .map(s -> getAvroSchema(s, rowType))
+ .orElse(AvroSchemaConverter.convertToSchema(rowType));
+ final TypeInformation<RowData> rowDataTypeInfo =
+ context.createTypeInformation(producedDataType);
+ return new AvroRowDataDeserializationSchema(
+ ApicurioRegistryAvroDeserializationSchema.forGeneric(
+ schema, schemaRegistryURL, optionalPropertiesMap),
+ AvroToRowDataConverters.createRowConverter(rowType),
+ rowDataTypeInfo);
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @Override
+ public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+ String schemaRegistryURL = formatOptions.get(URL);
+ Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+
+ return new EncodingFormat<SerializationSchema<RowData>>() {
+ @Override
+ public SerializationSchema<RowData> createRuntimeEncoder(
+ DynamicTableSink.Context context, DataType consumedDataType) {
+ final RowType rowType = (RowType) consumedDataType.getLogicalType();
+ final Schema schema =
+ schemaString
+ .map(s -> getAvroSchema(s, rowType))
+ .orElse(AvroSchemaConverter.convertToSchema(rowType));
+ return new AvroRowDataSerializationSchema(
+ rowType,
+ ApicurioRegistryAvroSerializationSchema.forGeneric(
+ schema, schemaRegistryURL, optionalPropertiesMap),
+ RowDataToAvroConverters.createConverter(rowType));
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @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() {
+ return getOptionalOptions();
+ }
+
+ protected static Set<ConfigOption<?>> getOptionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(USE_HEADERS);
+ options.add(USE_GLOBALID);
+ options.add(SCHEMA);
+ options.add(REGISTERED_ARTIFACT_NAME);
+ options.add(REGISTERED_ARTIFACT_DESCRIPTION);
+ options.add(REGISTERED_ARTIFACT_ID);
+ options.add(REGISTERED_ARTIFACT_VERSION);
+ 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_USERID);
+ options.add(BASIC_AUTH_CREDENTIALS_PASSWORD);
+ options.add(OIDC_AUTH_URL);
+ options.add(OIDC_AUTH_CLIENT_ID);
+ options.add(OIDC_AUTH_CLIENT_SECRET);
+ options.add(OIDC_AUTH_SCOPE);
+ options.add(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION);
+
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> forwardOptions() {
+ return Stream.of(
+ URL,
+ USE_HEADERS,
+ USE_GLOBALID,
+ SCHEMA,
+ REGISTERED_ARTIFACT_NAME,
+ REGISTERED_ARTIFACT_DESCRIPTION,
+ REGISTERED_ARTIFACT_ID,
+ REGISTERED_ARTIFACT_VERSION,
+ PROPERTIES,
+ SSL_KEYSTORE_LOCATION,
+ SSL_KEYSTORE_PASSWORD,
+ SSL_TRUSTSTORE_LOCATION,
+ SSL_TRUSTSTORE_PASSWORD,
+ BASIC_AUTH_CREDENTIALS_USERID,
+ BASIC_AUTH_CREDENTIALS_PASSWORD,
+ OIDC_AUTH_URL,
+ OIDC_AUTH_CLIENT_ID,
+ OIDC_AUTH_CLIENT_SECRET,
+ OIDC_AUTH_SCOPE,
+ OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)
+ .collect(Collectors.toSet());
+ }
+
+ public static @Nullable Map<String, Object> buildOptionalPropertiesMap(
+ ReadableConfig formatOptions) {
+ final Map<String, Object> properties = new HashMap<>();
+
+ Set<ConfigOption<?>> configOptionsArray = getOptionalOptions();
+
+ for (ConfigOption configOption : configOptionsArray) {
+ formatOptions
+ .getOptional(configOption)
+ .ifPresent(v -> properties.put(configOption.key(), v));
+ // when there is no value we would like to set the default as the value.
+ // at java 11 we can use the orElse or isEmpty , but we are java 8
+ String configOptionKey = configOption.key();
+ if (properties.get(configOptionKey) == null && configOption.hasDefaultValue()) {
+ properties.put(configOptionKey, configOption.defaultValue());
+ }
+ }
+ return properties;
+ }
+
+ private static Schema getAvroSchema(String schemaString, RowType rowType) {
+ LogicalType convertedDataType =
+ AvroSchemaConverter.convertToDataType(schemaString).getLogicalType();
+
+ if (convertedDataType.isNullable()) {
+ convertedDataType = convertedDataType.copy(false);
+ }
+
+ if (!convertedDataType.equals(rowType)) {
+ throw new IllegalArgumentException(
+ format(
+ "Schema provided for '%s' format does not match the table schema: %s",
+ IDENTIFIER, schemaString));
+ }
+
+ return new Parser().parse(schemaString);
Review Comment:
Can this throw errors? I'd expect a parse error or equivalent checked exception?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
Review Comment:
Is there an apicurio version check needed here? Is this an API that only exists on some apicurio versions?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
+ if (artifactReferenceList == null || artifactReferenceList.isEmpty()) {
+ throw new IOException(
+ "Could not find any schemas in the Apicurio Registry with content ID "
+ + schemaId);
+ }
+ long latestVersion = 0L;
+ String artifactId = "";
+ String groupId = "";
+ for (ArtifactReference artifactReference : artifactReferenceList) {
+ long version = Long.valueOf(artifactReference.getVersion());
+ // TODO check Apicurio code to see how it checks for the latest version
+ if (version > latestVersion) {
+ latestVersion = version;
+ artifactId = artifactReference.getArtifactId();
+ groupId = artifactReference.getGroupId();
+ }
+ }
+ VersionMetaData artifactVersion =
+ registryClient.getArtifactVersionMetaData(
+ groupId, artifactId, latestVersion + "");
+ schemaId = artifactVersion.getGlobalId();
+ }
+ // get the schema in canonical form with references dereferenced
+ InputStream schemaInputStream = registryClient.getContentByGlobalId(schemaId, true, true);
+ Schema schema = new Schema.Parser().parse(schemaInputStream);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ methodName
+ + " got schema "
+ + schema
+ + " using "
+ + (useGlobalId ? "global" : "content")
+ + " ID "
+ + schemaId);
+ }
+ return schema;
+ }
+
+ /**
+ * Get the schema id.
+ *
+ * @param in the Kafka body as an input stream
+ * @param additionalParameters additional Pararamters containing the headers and whetehr we are
+ * processing a key or value
+ * @param useGlobalID hen true globalIDs are used, otherwise contentIDS
+ * @return the id of the schema
+ * @throws IOException error occurred
+ */
+ protected long getSchemaId(
+ InputStream in, Map<String, Object> additionalParameters, boolean useGlobalID)
Review Comment:
Rather than passing in `additionalParameters`, is it cleaner to add a few parameters for `isKey`, `useHeaders`, and `Map<String,Object> headers` - so that this method doesn't have to know how to retrieve them from the nested map?
##########
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RegistryAvroSerializationSchema.java:
##########
@@ -107,17 +108,37 @@ public static RegistryAvroSerializationSchema<GenericRecord> forGeneric(
}
@Override
- public byte[] serialize(T object) {
- checkAvroInitialized();
+ public byte[] serialize(T element) {
+ return serializeWithAdditionalProperties(element, null, null);
+ }
+ @Override
+ public byte[] serializeWithAdditionalProperties(
+ T object,
+ Map<String, Object> inputAdditionalProperties,
+ Map<String, Object> outputAdditionalProperties) {
+ checkAvroInitialized();
if (object == null) {
return null;
} else {
try {
ByteArrayOutputStream outputStream = getOutputStream();
outputStream.reset();
Encoder encoder = getEncoder();
- schemaCoder.writeSchema(getSchema(), outputStream);
+ if (inputAdditionalProperties == null) {
Review Comment:
Why are we failing here?
Why are we not checking that outputProperties is not null?
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCodergetSchemaTest.java:
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for schemacoder. */
+public class ApicurioSchemaRegistryCodergetSchemaTest {
+
+ @ParameterizedTest
+ @MethodSource("configProvider")
+ public void getSchemaId(TestSpec testSpec) {
+ try {
+
+ if (testSpec.extraConfigOptions == null) {
+ testSpec.extraConfigOptions = new HashMap<>();
+ testSpec.extraConfigOptions.put(USE_GLOBALID.key(), true);
+ testSpec.extraConfigOptions.put(USE_HEADERS.key(), true);
+ }
+
+ // get options map
+ Map<String, Object> registryConfigs =
+ getConfig(testSpec.isKey, testSpec.extraConfigOptions);
+ // this is just to have something in the stream
+ String str = readFile("src/test/resources/simple1.avro", StandardCharsets.UTF_8);
+
+ byte[] schemaBytes = str.getBytes(StandardCharsets.UTF_8);
+ InputStream in;
+ if (testSpec.byteArray == null) {
+ in = new ByteArrayInputStream(schemaBytes);
+ } else {
+ // the byteArray is the magic byte and schema id that should be at the start of the
+ // body
+ byte[] combined = new byte[testSpec.byteArray.length + schemaBytes.length];
+ ByteBuffer byteBuffer = ByteBuffer.wrap(combined);
+ byteBuffer.put(testSpec.byteArray);
+ byteBuffer.put(schemaBytes);
+ combined = byteBuffer.array();
+ in = new ByteArrayInputStream(combined);
+ }
+
+ ApicurioSchemaRegistryCoder apicurioSchemaRegistryCoder =
Review Comment:
Worth some tests that provide an empty map for `registryConfigs` to tease out the null pointers mentioned above.
I appreciate that the config itself is built with defaults, but I don't know if it's safe to assume that someone will only call the constructor from the default config?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
+ if (artifactReferenceList == null || artifactReferenceList.isEmpty()) {
+ throw new IOException(
+ "Could not find any schemas in the Apicurio Registry with content ID "
+ + schemaId);
+ }
+ long latestVersion = 0L;
+ String artifactId = "";
+ String groupId = "";
+ for (ArtifactReference artifactReference : artifactReferenceList) {
+ long version = Long.valueOf(artifactReference.getVersion());
+ // TODO check Apicurio code to see how it checks for the latest version
+ if (version > latestVersion) {
+ latestVersion = version;
+ artifactId = artifactReference.getArtifactId();
+ groupId = artifactReference.getGroupId();
+ }
+ }
+ VersionMetaData artifactVersion =
+ registryClient.getArtifactVersionMetaData(
+ groupId, artifactId, latestVersion + "");
+ schemaId = artifactVersion.getGlobalId();
+ }
+ // get the schema in canonical form with references dereferenced
+ InputStream schemaInputStream = registryClient.getContentByGlobalId(schemaId, true, true);
+ Schema schema = new Schema.Parser().parse(schemaInputStream);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ methodName
+ + " got schema "
+ + schema
+ + " using "
+ + (useGlobalId ? "global" : "content")
+ + " ID "
+ + schemaId);
+ }
+ return schema;
+ }
+
+ /**
+ * Get the schema id.
+ *
+ * @param in the Kafka body as an input stream
+ * @param additionalParameters additional Pararamters containing the headers and whetehr we are
+ * processing a key or value
+ * @param useGlobalID hen true globalIDs are used, otherwise contentIDS
+ * @return the id of the schema
+ * @throws IOException error occurred
+ */
+ protected long getSchemaId(
+ InputStream in, Map<String, Object> additionalParameters, boolean useGlobalID)
+ throws IOException {
+ long schemaId = 0L;
+ Map<String, Object> headers = (Map<String, Object>) additionalParameters.get(HEADERS);
Review Comment:
Same issue here re null pointers on casting
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/AvroApicurioFormatOptions.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import io.apicurio.registry.serde.SerdeConfig;
+
+import java.time.Duration;
+import java.util.Map;
+
+/** Options for Schema Registry Avro format. */
+@PublicEvolving
+public class AvroApicurioFormatOptions {
+
+ public static final ConfigOption<String> URL =
+ ConfigOptions.key("url")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The URL of the Apicurio Registry to fetch/register schemas.");
+
+ public static final ConfigOption<Boolean> USE_HEADERS =
+ ConfigOptions.key("use-headers")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Used by serializers and deserializers. Configures the identifier for artifacts. \n"
+ + "true means use global ID, false means use content ID.\n"
+ + "Instructs the serializer to write the specified ID to Kafka,\n"
+ + "and instructs the deserializer to use this ID to find the schema.");
+
+ public static final ConfigOption<Boolean> USE_GLOBALID =
+ ConfigOptions.key("use-globalid")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Configures to read/write the artifact identifier to "
+ + "Kafka message headers instead of in the message payload.");
+
+ public static final ConfigOption<String> GROUP_ID =
+ ConfigOptions.key("groupId")
+ .stringType()
+ .defaultValue("default")
+ .withDescription("GroupId Used by deserializers.");
+
+ public static final ConfigOption<String> SCHEMA =
+ ConfigOptions.key("schema")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The schema registered or to be registered in the Apicurio Registry. "
+ + "If no schema is provided Flink converts the table schema to avro schema. "
+ + "The schema provided must match the table schema ('avro-apicurio').");
Review Comment:
https://github.com/apache/flink/pull/24715/files#diff-d8f33c08f59e3f895d0fa092cb551c8e3378f3a20d19c4845384377909869057R268 - is doing an equality check, not a compatibility check.
Which currently reads like:
- If you do not provide a schema, the writer schema is generated from the table
- If you do provide a schema, it must match the row type (which makes the ability to provide a schema moot)
If this is for providing the writer schema, I would expect the `.equals()` to be a compatibility check instead.
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
+ if (artifactReferenceList == null || artifactReferenceList.isEmpty()) {
+ throw new IOException(
+ "Could not find any schemas in the Apicurio Registry with content ID "
+ + schemaId);
+ }
+ long latestVersion = 0L;
+ String artifactId = "";
+ String groupId = "";
+ for (ArtifactReference artifactReference : artifactReferenceList) {
+ long version = Long.valueOf(artifactReference.getVersion());
+ // TODO check Apicurio code to see how it checks for the latest version
+ if (version > latestVersion) {
+ latestVersion = version;
+ artifactId = artifactReference.getArtifactId();
+ groupId = artifactReference.getGroupId();
+ }
+ }
+ VersionMetaData artifactVersion =
+ registryClient.getArtifactVersionMetaData(
+ groupId, artifactId, latestVersion + "");
+ schemaId = artifactVersion.getGlobalId();
+ }
+ // get the schema in canonical form with references dereferenced
+ InputStream schemaInputStream = registryClient.getContentByGlobalId(schemaId, true, true);
+ Schema schema = new Schema.Parser().parse(schemaInputStream);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ methodName
+ + " got schema "
+ + schema
+ + " using "
+ + (useGlobalId ? "global" : "content")
+ + " ID "
+ + schemaId);
+ }
+ return schema;
+ }
+
+ /**
+ * Get the schema id.
+ *
+ * @param in the Kafka body as an input stream
+ * @param additionalParameters additional Pararamters containing the headers and whetehr we are
+ * processing a key or value
+ * @param useGlobalID hen true globalIDs are used, otherwise contentIDS
+ * @return the id of the schema
+ * @throws IOException error occurred
+ */
+ protected long getSchemaId(
+ InputStream in, Map<String, Object> additionalParameters, boolean useGlobalID)
+ throws IOException {
+ long schemaId = 0L;
+ Map<String, Object> headers = (Map<String, Object>) additionalParameters.get(HEADERS);
+ boolean isKey = (boolean) additionalParameters.get(IS_KEY);
+ boolean useHeaders = (boolean) configs.get(USE_HEADERS.key());
Review Comment:
null pointer if `.get()` returns null
##########
flink-core/src/main/java/org/apache/flink/api/common/serialization/DeserializationSchema.java:
##########
@@ -94,6 +95,31 @@ default void deserialize(byte[] message, Collector<T> out) throws IOException {
*/
boolean isEndOfStream(T nextElement);
+ /**
+ * Deserializes the byte message with input Additional Properties.
+ *
+ * @param message The message, as a byte array.
+ * @param inputAdditionalProperties inputAdditionalProperties map of input Additional Properties
+ * that can be used for deserialization. Override this method to make use of the
+ * inputAdditionalProperties,
+ * @return The deserialized message as an object (null if the message cannot be deserialized).
+ */
+ @PublicEvolving
+ default T deserializeWithAdditionalProperties(
+ byte[] message, Map<String, Object> inputAdditionalProperties) throws IOException {
+ return deserialize(message);
+ }
+
+ @PublicEvolving
+ default void deserializeWithAdditionalProperties(
+ byte[] message, Map<String, Object> inputAdditionalProperties, Collector<T> out)
+ throws IOException {
+ T deserialize = deserializeWithAdditionalProperties(message, inputAdditionalProperties);
Review Comment:
Why does this default method not just call `deserialize(byte[] message, Collector<T> out)`?
Is it a future proof to make sure that if `deserializeWithAdditionalProperties` default changes, we pick it up here?
##########
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/SchemaCoder.java:
##########
@@ -34,6 +35,20 @@ public interface SchemaCoder {
void writeSchema(Schema schema, OutputStream out) throws IOException;
+ default void writeSchema(
+ Schema schema,
+ OutputStream out,
+ Map<String, Object> inputProperties,
+ Map<String, Object> outputProperties)
+ throws IOException {
+ throw new RuntimeException("writeSchema passing headers should be overridden.");
Review Comment:
why is this not calling `writeSchema(schema, out)` like you do with `readSchemaWithAdditionalParameters`?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
+ if (artifactReferenceList == null || artifactReferenceList.isEmpty()) {
+ throw new IOException(
+ "Could not find any schemas in the Apicurio Registry with content ID "
+ + schemaId);
+ }
+ long latestVersion = 0L;
+ String artifactId = "";
+ String groupId = "";
+ for (ArtifactReference artifactReference : artifactReferenceList) {
+ long version = Long.valueOf(artifactReference.getVersion());
+ // TODO check Apicurio code to see how it checks for the latest version
+ if (version > latestVersion) {
+ latestVersion = version;
+ artifactId = artifactReference.getArtifactId();
+ groupId = artifactReference.getGroupId();
+ }
+ }
+ VersionMetaData artifactVersion =
+ registryClient.getArtifactVersionMetaData(
+ groupId, artifactId, latestVersion + "");
+ schemaId = artifactVersion.getGlobalId();
+ }
+ // get the schema in canonical form with references dereferenced
+ InputStream schemaInputStream = registryClient.getContentByGlobalId(schemaId, true, true);
+ Schema schema = new Schema.Parser().parse(schemaInputStream);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ methodName
+ + " got schema "
+ + schema
+ + " using "
+ + (useGlobalId ? "global" : "content")
+ + " ID "
+ + schemaId);
+ }
+ return schema;
+ }
+
+ /**
+ * Get the schema id.
+ *
+ * @param in the Kafka body as an input stream
+ * @param additionalParameters additional Pararamters containing the headers and whetehr we are
+ * processing a key or value
+ * @param useGlobalID hen true globalIDs are used, otherwise contentIDS
+ * @return the id of the schema
+ * @throws IOException error occurred
+ */
+ protected long getSchemaId(
+ InputStream in, Map<String, Object> additionalParameters, boolean useGlobalID)
+ throws IOException {
+ long schemaId = 0L;
+ Map<String, Object> headers = (Map<String, Object>) additionalParameters.get(HEADERS);
Review Comment:
Why cast to `<String, Object>` when everything in the method accesses it as `<String, byte[]>` with extra casting?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
Review Comment:
potential null pointer here if the map doesn't contain the key
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioRegistryAvroFormatFactory.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.formats.avro.AvroRowDataDeserializationSchema;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
+import org.apache.flink.formats.avro.AvroToRowDataConverters;
+import org.apache.flink.formats.avro.RowDataToAvroConverters;
+import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.Projection;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.format.ProjectableDecodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.GenericRowData;
+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 org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.avro.Schema;
+import org.apache.avro.Schema.Parser;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.lang.String.format;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_SECRET;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_SCOPE;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_DESCRIPTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_NAME;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_VERSION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SCHEMA;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Apicurio avro format. */
+@Internal
+public class ApicurioRegistryAvroFormatFactory
+ implements DeserializationFormatFactory, SerializationFormatFactory {
+
+ // used this key rather than apicurio-avro (the format name) to be consistent with Confluent.
+ public static final String IDENTIFIER = "avro-apicurio";
+
+ @Override
+ public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+ String schemaRegistryURL = formatOptions.get(URL);
+ Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+ return new ProjectableDecodingFormat<DeserializationSchema<RowData>>() {
+ // --------------------------------------------------------------------------------------------
+ // Mutable attributes
+ // --------------------------------------------------------------------------------------------
+
+ private List<String> metadataKeys;
+
+ @Override
+ public DeserializationSchema<RowData> createRuntimeDecoder(
+ DynamicTableSource.Context context,
+ DataType producedDataType,
+ int[][] projections) {
+ producedDataType = Projection.of(projections).project(producedDataType);
+ final RowType rowType = (RowType) producedDataType.getLogicalType();
+ // schema has been supplied
+ final Schema schema =
+ schemaString
+ .map(s -> getAvroSchema(s, rowType))
+ .orElse(AvroSchemaConverter.convertToSchema(rowType));
+ final TypeInformation<RowData> rowDataTypeInfo =
+ context.createTypeInformation(producedDataType);
+ return new AvroRowDataDeserializationSchema(
+ ApicurioRegistryAvroDeserializationSchema.forGeneric(
+ schema, schemaRegistryURL, optionalPropertiesMap),
+ AvroToRowDataConverters.createRowConverter(rowType),
+ rowDataTypeInfo);
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @Override
+ public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+ String schemaRegistryURL = formatOptions.get(URL);
+ Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+
+ return new EncodingFormat<SerializationSchema<RowData>>() {
+ @Override
+ public SerializationSchema<RowData> createRuntimeEncoder(
+ DynamicTableSink.Context context, DataType consumedDataType) {
+ final RowType rowType = (RowType) consumedDataType.getLogicalType();
+ final Schema schema =
+ schemaString
+ .map(s -> getAvroSchema(s, rowType))
+ .orElse(AvroSchemaConverter.convertToSchema(rowType));
+ return new AvroRowDataSerializationSchema(
+ rowType,
+ ApicurioRegistryAvroSerializationSchema.forGeneric(
+ schema, schemaRegistryURL, optionalPropertiesMap),
+ RowDataToAvroConverters.createConverter(rowType));
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @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() {
+ return getOptionalOptions();
+ }
+
+ protected static Set<ConfigOption<?>> getOptionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(USE_HEADERS);
+ options.add(USE_GLOBALID);
+ options.add(SCHEMA);
+ options.add(REGISTERED_ARTIFACT_NAME);
+ options.add(REGISTERED_ARTIFACT_DESCRIPTION);
+ options.add(REGISTERED_ARTIFACT_ID);
+ options.add(REGISTERED_ARTIFACT_VERSION);
+ 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_USERID);
+ options.add(BASIC_AUTH_CREDENTIALS_PASSWORD);
+ options.add(OIDC_AUTH_URL);
+ options.add(OIDC_AUTH_CLIENT_ID);
+ options.add(OIDC_AUTH_CLIENT_SECRET);
+ options.add(OIDC_AUTH_SCOPE);
+ options.add(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION);
+
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> forwardOptions() {
+ return Stream.of(
+ URL,
+ USE_HEADERS,
+ USE_GLOBALID,
+ SCHEMA,
+ REGISTERED_ARTIFACT_NAME,
+ REGISTERED_ARTIFACT_DESCRIPTION,
+ REGISTERED_ARTIFACT_ID,
+ REGISTERED_ARTIFACT_VERSION,
+ PROPERTIES,
+ SSL_KEYSTORE_LOCATION,
+ SSL_KEYSTORE_PASSWORD,
+ SSL_TRUSTSTORE_LOCATION,
+ SSL_TRUSTSTORE_PASSWORD,
+ BASIC_AUTH_CREDENTIALS_USERID,
+ BASIC_AUTH_CREDENTIALS_PASSWORD,
+ OIDC_AUTH_URL,
+ OIDC_AUTH_CLIENT_ID,
+ OIDC_AUTH_CLIENT_SECRET,
+ OIDC_AUTH_SCOPE,
+ OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)
+ .collect(Collectors.toSet());
+ }
+
+ public static @Nullable Map<String, Object> buildOptionalPropertiesMap(
+ ReadableConfig formatOptions) {
+ final Map<String, Object> properties = new HashMap<>();
+
+ Set<ConfigOption<?>> configOptionsArray = getOptionalOptions();
+
+ for (ConfigOption configOption : configOptionsArray) {
+ formatOptions
+ .getOptional(configOption)
+ .ifPresent(v -> properties.put(configOption.key(), v));
+ // when there is no value we would like to set the default as the value.
+ // at java 11 we can use the orElse or isEmpty , but we are java 8
+ String configOptionKey = configOption.key();
+ if (properties.get(configOptionKey) == null && configOption.hasDefaultValue()) {
+ properties.put(configOptionKey, configOption.defaultValue());
+ }
+ }
+ return properties;
+ }
+
+ private static Schema getAvroSchema(String schemaString, RowType rowType) {
+ LogicalType convertedDataType =
Review Comment:
Can this throw errors? I'd expect a parse error or equivalent checked exception?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
+ if (artifactReferenceList == null || artifactReferenceList.isEmpty()) {
+ throw new IOException(
+ "Could not find any schemas in the Apicurio Registry with content ID "
+ + schemaId);
+ }
+ long latestVersion = 0L;
+ String artifactId = "";
+ String groupId = "";
+ for (ArtifactReference artifactReference : artifactReferenceList) {
+ long version = Long.valueOf(artifactReference.getVersion());
+ // TODO check Apicurio code to see how it checks for the latest version
+ if (version > latestVersion) {
+ latestVersion = version;
+ artifactId = artifactReference.getArtifactId();
+ groupId = artifactReference.getGroupId();
+ }
+ }
+ VersionMetaData artifactVersion =
+ registryClient.getArtifactVersionMetaData(
+ groupId, artifactId, latestVersion + "");
+ schemaId = artifactVersion.getGlobalId();
+ }
+ // get the schema in canonical form with references dereferenced
+ InputStream schemaInputStream = registryClient.getContentByGlobalId(schemaId, true, true);
+ Schema schema = new Schema.Parser().parse(schemaInputStream);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ methodName
+ + " got schema "
+ + schema
+ + " using "
+ + (useGlobalId ? "global" : "content")
+ + " ID "
+ + schemaId);
+ }
+ return schema;
+ }
+
+ /**
+ * Get the schema id.
+ *
+ * @param in the Kafka body as an input stream
+ * @param additionalParameters additional Pararamters containing the headers and whetehr we are
+ * processing a key or value
+ * @param useGlobalID hen true globalIDs are used, otherwise contentIDS
+ * @return the id of the schema
+ * @throws IOException error occurred
+ */
+ protected long getSchemaId(
+ InputStream in, Map<String, Object> additionalParameters, boolean useGlobalID)
+ throws IOException {
+ long schemaId = 0L;
+ Map<String, Object> headers = (Map<String, Object>) additionalParameters.get(HEADERS);
+ boolean isKey = (boolean) additionalParameters.get(IS_KEY);
+ boolean useHeaders = (boolean) configs.get(USE_HEADERS.key());
+
+ if (useHeaders) {
+ String globalIDHeaderName =
+ isKey ? APICURIO_KEY_GLOBAL_ID_HEADER : APICURIO_VALUE_GLOBAL_ID_HEADER;
+ String contentIDHeaderName =
+ isKey ? APICURIO_KEY_CONTENT_ID_HEADER : APICURIO_VALUE_CONTENT_ID_HEADER;
+ // get from headers
+ if (headers.get(globalIDHeaderName) != null) {
+ if (!useGlobalID) {
+ throw new IOException(
+ "The message has a global ID header "
+ + globalIDHeaderName
+ + ", but the configuration is for content ID.\n"
+ + "Header names received: \n"
+ + headers.keySet().stream().collect(Collectors.joining(",")));
+ }
+ byte[] globalIDByteArray = (byte[]) headers.get(globalIDHeaderName);
+ schemaId = bytesToLong(globalIDByteArray);
+ } else if (headers.get(contentIDHeaderName) != null) {
+ if (useGlobalID) {
+ throw new IOException(
+ "The message has a content ID header "
+ + contentIDHeaderName
+ + "\n"
+ + ", but the configuration is for global ID.\n"
+ + "Header names received: "
+ + headers.keySet().stream().collect(Collectors.joining(",")));
+ }
+ byte[] contentIDByteArray = (byte[]) headers.get(contentIDHeaderName);
+ schemaId = bytesToLong(contentIDByteArray);
+ } else {
+ String headerName = useGlobalID ? globalIDHeaderName : contentIDHeaderName;
+ throw new IOException(
+ "The format was configured to enable headers, but the expected Kafka header "
+ + headerName
+ + " is not present.\n "
+ + "Header names received: "
+ + headers.keySet().stream().collect(Collectors.joining(","))
+ + "\n "
+ + "Ensure that the kafka message was created by an Apicurio client. Check that it is "
+ + "sending the appropriate ID in the header, if the message is sending the ID in the payload,\n"
+ + " review and amend the format configuration options "
+ + USE_HEADERS
+ + " and "
+ + USE_GLOBALID);
+ }
+ } else {
+ // the id is in the payload
+ DataInputStream dataInputStream = new DataInputStream(in);
+ if (dataInputStream.readByte() != 0) {
+ throw new IOException(
+ "Unknown data format. Magic number was not found. "
+ + USE_HEADERS
+ + "= false"
+ + " configs="
+ + configs);
+ }
+ schemaId = dataInputStream.readLong();
+ }
+ return schemaId;
+ }
+
+ @Override
+ public void writeSchema(Schema schema, OutputStream out) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector is being used, upgrade the Kafka connector.");
+ }
+
+ /**
+ * The Avro schema will be written depending on the configuration.
+ *
+ * @param schema Avro Schema to write into the output stream,
+ * @param out the output stream where the schema is serialized.
+ * @param inputProperties properties containing the topic name
+ * @param outputProperties headers to populate
+ * @throws IOException Exception has occurred
+ */
+ @Override
+ public void writeSchema(
+ Schema schema,
+ OutputStream out,
+ Map<String, Object> inputProperties,
+ Map<String, Object> outputProperties)
+ throws IOException {
+ String methodName = "writeSchema with additional properties";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+
+ boolean useHeaders = (boolean) configs.get(USE_HEADERS.key());
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+ boolean isKey = (boolean) inputProperties.get(IS_KEY);
+
+ String groupId = (String) configs.get(AvroApicurioFormatOptions.GROUP_ID.key());
+ Object artifactId = configs.get(AvroApicurioFormatOptions.REGISTERED_ARTIFACT_ID.key());
+ Object artifactName = configs.get(AvroApicurioFormatOptions.REGISTERED_ARTIFACT_NAME.key());
+ if (inputProperties.get(TOPIC_NAME) == null) {
+ String configOptionName = null;
+ if (artifactId == null) {
+ configOptionName = "artifactId";
+ }
+ if (artifactName == null) {
+ configOptionName = "artifactName";
+ }
+ if (configOptionName != null) {
+ throw new IOException(
+ "No topic name was configured, and it is required to provide a default for "
+ + configOptionName);
+ }
+ }
+ String defaultArtifactName =
+ (String) inputProperties.get(TOPIC_NAME + (isKey ? "-key" : "-value"));
+ artifactId = (artifactId == null) ? defaultArtifactName : artifactId;
+ artifactName = (artifactName == null) ? defaultArtifactName : artifactName;
+
+ // config option has a default so no need to check for null
+ String artifactDescription =
+ (String)
Review Comment:
More occurrances of null pointers when casting from `.get()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft pr for initial review of apicurio [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24715:
URL: https://github.com/apache/flink/pull/24715#issuecomment-2075321251
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0fdfcddfaed3a5fb73e6bbb01059073f90e5e587",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0fdfcddfaed3a5fb73e6bbb01059073f90e5e587",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0fdfcddfaed3a5fb73e6bbb01059073f90e5e587 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] FLINK-35311 Draft pr for initial review of apicurio [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl closed pull request #24715: FLINK-35311 Draft pr for initial review of apicurio
URL: https://github.com/apache/flink/pull/24715
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Draft pr for initial review of apicurio [flink]
Posted by "mnuttall (via GitHub)" <gi...@apache.org>.
mnuttall commented on code in PR #24715:
URL: https://github.com/apache/flink/pull/24715#discussion_r1605104247
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioRegistryAvroFormatFactory.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.formats.avro.AvroRowDataDeserializationSchema;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
+import org.apache.flink.formats.avro.AvroToRowDataConverters;
+import org.apache.flink.formats.avro.RowDataToAvroConverters;
+import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.Projection;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.format.ProjectableDecodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.GenericRowData;
+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 org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.avro.Schema;
+import org.apache.avro.Schema.Parser;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.lang.String.format;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_SECRET;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_SCOPE;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_DESCRIPTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_NAME;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_VERSION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SCHEMA;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Apicurio avro format. */
+@Internal
+public class ApicurioRegistryAvroFormatFactory
+ implements DeserializationFormatFactory, SerializationFormatFactory {
+
+ // used this key rather than apicurio-avro (the format name) to be consistent with Confluent.
+ public static final String IDENTIFIER = "avro-apicurio";
+
+ @Override
+ public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+ String schemaRegistryURL = formatOptions.get(URL);
+ Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+ return new ProjectableDecodingFormat<DeserializationSchema<RowData>>() {
+ // --------------------------------------------------------------------------------------------
+ // Mutable attributes
+ // --------------------------------------------------------------------------------------------
+
+ private List<String> metadataKeys;
+
+ @Override
+ public DeserializationSchema<RowData> createRuntimeDecoder(
+ DynamicTableSource.Context context,
+ DataType producedDataType,
+ int[][] projections) {
+ producedDataType = Projection.of(projections).project(producedDataType);
+ final RowType rowType = (RowType) producedDataType.getLogicalType();
+ // schema has been supplied
+ final Schema schema =
+ schemaString
+ .map(s -> getAvroSchema(s, rowType))
+ .orElse(AvroSchemaConverter.convertToSchema(rowType));
+ final TypeInformation<RowData> rowDataTypeInfo =
+ context.createTypeInformation(producedDataType);
+ return new AvroRowDataDeserializationSchema(
+ ApicurioRegistryAvroDeserializationSchema.forGeneric(
+ schema, schemaRegistryURL, optionalPropertiesMap),
+ AvroToRowDataConverters.createRowConverter(rowType),
+ rowDataTypeInfo);
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @Override
+ public EncodingFormat<SerializationSchema<RowData>> createEncodingFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+ String schemaRegistryURL = formatOptions.get(URL);
+ Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+
+ return new EncodingFormat<SerializationSchema<RowData>>() {
+ @Override
+ public SerializationSchema<RowData> createRuntimeEncoder(
+ DynamicTableSink.Context context, DataType consumedDataType) {
+ final RowType rowType = (RowType) consumedDataType.getLogicalType();
+ final Schema schema =
+ schemaString
+ .map(s -> getAvroSchema(s, rowType))
+ .orElse(AvroSchemaConverter.convertToSchema(rowType));
+ return new AvroRowDataSerializationSchema(
+ rowType,
+ ApicurioRegistryAvroSerializationSchema.forGeneric(
+ schema, schemaRegistryURL, optionalPropertiesMap),
+ RowDataToAvroConverters.createConverter(rowType));
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @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() {
+ return getOptionalOptions();
+ }
+
+ protected static Set<ConfigOption<?>> getOptionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(USE_HEADERS);
+ options.add(USE_GLOBALID);
+ options.add(SCHEMA);
+ options.add(REGISTERED_ARTIFACT_NAME);
+ options.add(REGISTERED_ARTIFACT_DESCRIPTION);
+ options.add(REGISTERED_ARTIFACT_ID);
+ options.add(REGISTERED_ARTIFACT_VERSION);
+ 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_USERID);
+ options.add(BASIC_AUTH_CREDENTIALS_PASSWORD);
+ options.add(OIDC_AUTH_URL);
+ options.add(OIDC_AUTH_CLIENT_ID);
+ options.add(OIDC_AUTH_CLIENT_SECRET);
+ options.add(OIDC_AUTH_SCOPE);
+ options.add(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION);
+
+ return options;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> forwardOptions() {
+ return Stream.of(
+ URL,
+ USE_HEADERS,
+ USE_GLOBALID,
+ SCHEMA,
+ REGISTERED_ARTIFACT_NAME,
+ REGISTERED_ARTIFACT_DESCRIPTION,
+ REGISTERED_ARTIFACT_ID,
+ REGISTERED_ARTIFACT_VERSION,
+ PROPERTIES,
+ SSL_KEYSTORE_LOCATION,
+ SSL_KEYSTORE_PASSWORD,
+ SSL_TRUSTSTORE_LOCATION,
+ SSL_TRUSTSTORE_PASSWORD,
+ BASIC_AUTH_CREDENTIALS_USERID,
+ BASIC_AUTH_CREDENTIALS_PASSWORD,
+ OIDC_AUTH_URL,
+ OIDC_AUTH_CLIENT_ID,
+ OIDC_AUTH_CLIENT_SECRET,
+ OIDC_AUTH_SCOPE,
+ OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)
+ .collect(Collectors.toSet());
+ }
+
+ public static @Nullable Map<String, Object> buildOptionalPropertiesMap(
+ ReadableConfig formatOptions) {
+ final Map<String, Object> properties = new HashMap<>();
+
+ Set<ConfigOption<?>> configOptionsArray = getOptionalOptions();
+
+ for (ConfigOption configOption : configOptionsArray) {
+ formatOptions
+ .getOptional(configOption)
+ .ifPresent(v -> properties.put(configOption.key(), v));
+ // when there is no value we would like to set the default as the value.
+ // at java 11 we can use the orElse or isEmpty , but we are java 8
+ String configOptionKey = configOption.key();
+ if (properties.get(configOptionKey) == null && configOption.hasDefaultValue()) {
+ properties.put(configOptionKey, configOption.defaultValue());
+ }
+ }
+ return properties;
+ }
+
+ private static Schema getAvroSchema(String schemaString, RowType rowType) {
+ LogicalType convertedDataType =
+ AvroSchemaConverter.convertToDataType(schemaString).getLogicalType();
+
+ if (convertedDataType.isNullable()) {
Review Comment:
Do you make `convertedDataType` non-nullable because `rowType` is always non-nullable, and the `.equals()` below won't work otherwise? A comment here could be helpful.
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
+ if (artifactReferenceList == null || artifactReferenceList.isEmpty()) {
+ throw new IOException(
+ "Could not find any schemas in the Apicurio Registry with content ID "
+ + schemaId);
+ }
+ long latestVersion = 0L;
+ String artifactId = "";
+ String groupId = "";
+ for (ArtifactReference artifactReference : artifactReferenceList) {
+ long version = Long.valueOf(artifactReference.getVersion());
+ // TODO check Apicurio code to see how it checks for the latest version
Review Comment:
// TODO
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
Review Comment:
What's the connection between `readSchema()` and 'a Kafka connector that passes headers' ?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioRegistryAvroFormatFactory.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.formats.avro.AvroRowDataDeserializationSchema;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
+import org.apache.flink.formats.avro.AvroToRowDataConverters;
+import org.apache.flink.formats.avro.RowDataToAvroConverters;
+import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.Projection;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.format.EncodingFormat;
+import org.apache.flink.table.connector.format.ProjectableDecodingFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.GenericRowData;
+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 org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.avro.Schema;
+import org.apache.avro.Schema.Parser;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static java.lang.String.format;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_SECRET;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_SCOPE;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_DESCRIPTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_NAME;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.REGISTERED_ARTIFACT_VERSION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SCHEMA;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_KEYSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.URL;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Apicurio avro format. */
+@Internal
+public class ApicurioRegistryAvroFormatFactory
+ implements DeserializationFormatFactory, SerializationFormatFactory {
+
+ // used this key rather than apicurio-avro (the format name) to be consistent with Confluent.
+ public static final String IDENTIFIER = "avro-apicurio";
+
+ @Override
+ public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+
+ String schemaRegistryURL = formatOptions.get(URL);
+ Optional<String> schemaString = formatOptions.getOptional(SCHEMA);
+ Map<String, ?> optionalPropertiesMap = buildOptionalPropertiesMap(formatOptions);
+ return new ProjectableDecodingFormat<DeserializationSchema<RowData>>() {
+ // --------------------------------------------------------------------------------------------
Review Comment:
The comments in this method don't add very much and could probably be removed
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/AvroApicurioFormatOptions.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import io.apicurio.registry.serde.SerdeConfig;
+
+import java.time.Duration;
+import java.util.Map;
+
+/** Options for Schema Registry Avro format. */
+@PublicEvolving
+public class AvroApicurioFormatOptions {
+
+ public static final ConfigOption<String> URL =
+ ConfigOptions.key("url")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The URL of the Apicurio Registry to fetch/register schemas.");
+
+ public static final ConfigOption<Boolean> USE_HEADERS =
+ ConfigOptions.key("use-headers")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Used by serializers and deserializers. Configures the identifier for artifacts. \n"
+ + "true means use global ID, false means use content ID.\n"
+ + "Instructs the serializer to write the specified ID to Kafka,\n"
+ + "and instructs the deserializer to use this ID to find the schema.");
+
+ public static final ConfigOption<Boolean> USE_GLOBALID =
+ ConfigOptions.key("use-globalid")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Configures to read/write the artifact identifier to "
+ + "Kafka message headers instead of in the message payload.");
+
+ public static final ConfigOption<String> GROUP_ID =
+ ConfigOptions.key("groupId")
+ .stringType()
+ .defaultValue("default")
+ .withDescription("GroupId Used by deserializers.");
+
+ public static final ConfigOption<String> SCHEMA =
+ ConfigOptions.key("schema")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The schema registered or to be registered in the Apicurio Registry. "
+ + "If no schema is provided Flink converts the table schema to avro schema. "
+ + "The schema provided must match the table schema ('avro-apicurio').");
+ public static final ConfigOption<String> REGISTERED_ARTIFACT_ID =
+ ConfigOptions.key("artifactId")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "ArtifactId is used by serializers when registering a new Schemas to ensure that\n"
+ + "each serialization is associated with the same schema in the registry.\n"
+ + "It is often useful to include the topic name in the artifact id so\n"
+ + "the schema can be identified with its associated topic; the default is the topic name.");
+ public static final ConfigOption<String> REGISTERED_ARTIFACT_NAME =
+ ConfigOptions.key("artifactName")
+ .stringType()
+ .defaultValue("generated-schema")
+ .withDescription(
+ "The registered artifact name is used by serializers as the name of the schema being registered; \n"
+ + "the default is the topic name.");
+ public static final ConfigOption<String> REGISTERED_ARTIFACT_DESCRIPTION =
+ ConfigOptions.key("artifactDescription")
+ .stringType()
+ .defaultValue("Schema registered by Apache Flink.")
+ .withDescription(
+ "The registered schema description is used by serializers as the description of the schema being registered");
+ public static final ConfigOption<String> REGISTERED_ARTIFACT_VERSION =
+ ConfigOptions.key("artifactVersion")
+ .stringType()
+ .defaultValue("1")
+ .withDescription(
+ "The registered artifact version is used by serializers as the version of the schema being registered");
+
+ // --------------------------------------------------------------------------------------------
+ // Commonly used options maintained by Flink for convenience
+ // --------------------------------------------------------------------------------------------
+
+ public static final ConfigOption<String> SSL_KEYSTORE_LOCATION =
+ ConfigOptions.key(SerdeConfig.REQUEST_KEYSTORE_LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Location / File of SSL keystore");
+
+ public static final ConfigOption<String> SSL_KEYSTORE_PASSWORD =
+ ConfigOptions.key(SerdeConfig.REQUEST_KEYSTORE_PASSWORD)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Password for SSL keystore");
+ public static final ConfigOption<String> SSL_KEYSTORE_TYPE =
+ ConfigOptions.key(SerdeConfig.REQUEST_KEYSTORE_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Type for SSL truststore");
+
+ public static final ConfigOption<String> SSL_TRUSTSTORE_LOCATION =
+ ConfigOptions.key(SerdeConfig.REQUEST_TRUSTSTORE_LOCATION)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Location / File of SSL truststore");
+
+ public static final ConfigOption<String> SSL_TRUSTSTORE_PASSWORD =
+ ConfigOptions.key(SerdeConfig.REQUEST_TRUSTSTORE_PASSWORD)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Password for SSL truststore");
+
+ public static final ConfigOption<String> SSL_TRUSTSTORE_TYPE =
+ ConfigOptions.key(SerdeConfig.REQUEST_TRUSTSTORE_TYPE)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Type for SSL truststore");
+
+ public static final ConfigOption<String> BASIC_AUTH_CREDENTIALS_USERID =
+ ConfigOptions.key(SerdeConfig.AUTH_USERNAME)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Basic auth userid for Apicurio Registry");
+
+ public static final ConfigOption<String> BASIC_AUTH_CREDENTIALS_PASSWORD =
+ ConfigOptions.key(SerdeConfig.AUTH_PASSWORD)
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Basic auth password for Apicurio Registry");
+
+ public static final ConfigOption<String> OIDC_AUTH_URL =
+ ConfigOptions.key("OIDC_AUTH_URL")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("OIDC endpoint for Authorization.");
+
+ public static final ConfigOption<String> OIDC_AUTH_CLIENT_ID =
+ ConfigOptions.key("OIDC_AUTH_CLIENT_ID")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("OIDC Client ID for Apicurio Registry.");
+
+ public static final ConfigOption<String> OIDC_AUTH_CLIENT_SECRET =
+ ConfigOptions.key("OIDC_AUTH_CLIENT_SECRET")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("OIDC Client SECRET for Apicurio Registry.");
+
+ public static final ConfigOption<String> OIDC_AUTH_SCOPE =
+ ConfigOptions.key("OIDC_AUTH_SCOPE")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("OIDC scope for Apicurio Registry.");
+
+ public static final ConfigOption<Duration> OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION =
+ ConfigOptions.key("tokenExpirationReduction")
+ .durationType()
+ .noDefaultValue()
+ .withDescription("OIDC token expiry reduction for Apicurio Registry.");
+
+ // TODO the other SerdeConfig. options
Review Comment:
// TODO
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/CachedSchemaCoderProvider.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.client.RegistryClientFactory;
+import io.apicurio.rest.client.JdkHttpClient;
+import io.apicurio.rest.client.auth.Auth;
+import io.apicurio.rest.client.auth.BasicAuth;
+import io.apicurio.rest.client.auth.OidcAuth;
+import io.apicurio.rest.client.auth.exception.AuthErrorHandler;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_SECRET;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_URL;
+
+/**
+ * A {@link SchemaCoder.SchemaCoderProvider} that uses a cached schema registry client underneath.
+ */
+@Internal
+class CachedSchemaCoderProvider implements SchemaCoder.SchemaCoderProvider {
+
+ private static final long serialVersionUID = 8610401613495438381L;
+
+ private final String url;
+ private final int identityMapCapacity;
+ private final @Nullable Map<String, ?> configs;
+
+ CachedSchemaCoderProvider(String url, int identityMapCapacity) {
+ this(url, identityMapCapacity, null);
+ }
+
+ CachedSchemaCoderProvider(
+ String url, int identityMapCapacity, @Nullable Map<String, ?> configs) {
+ this.url = Objects.requireNonNull(url);
+ // TODO this does not seem to be used for Apicurio
Review Comment:
// TODO
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/CachedSchemaCoderProviderTest.java:
##########
@@ -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.
+ */
+
+package org.apache.flink.formats.avro.registry.apicurio;
+
+import org.junit.jupiter.api.Test;
+
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Tests for properties set by {@link ApicurioRegistryAvroFormatFactory} in {@link
+ * CachedSchemaCoderProvider}.
+ */
+class CachedSchemaCoderProviderTest {
+
+ @Test
+ void testThatSslIsNotInitializedForNoSslProperties() {
+ CachedSchemaCoderProvider provider = initCachedSchemaCoderProvider(new HashMap<>());
+ // SSLSocketFactory sslSocketFactory = getSslSocketFactoryFromProvider(provider);
Review Comment:
commented out code
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/CachedSchemaCoderProvider.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.client.RegistryClientFactory;
+import io.apicurio.rest.client.JdkHttpClient;
+import io.apicurio.rest.client.auth.Auth;
+import io.apicurio.rest.client.auth.BasicAuth;
+import io.apicurio.rest.client.auth.OidcAuth;
+import io.apicurio.rest.client.auth.exception.AuthErrorHandler;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_ID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_CLIENT_SECRET;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_URL;
+
+/**
+ * A {@link SchemaCoder.SchemaCoderProvider} that uses a cached schema registry client underneath.
+ */
+@Internal
+class CachedSchemaCoderProvider implements SchemaCoder.SchemaCoderProvider {
+
+ private static final long serialVersionUID = 8610401613495438381L;
+
+ private final String url;
+ private final int identityMapCapacity;
+ private final @Nullable Map<String, ?> configs;
+
+ CachedSchemaCoderProvider(String url, int identityMapCapacity) {
+ this(url, identityMapCapacity, null);
+ }
+
+ CachedSchemaCoderProvider(
+ String url, int identityMapCapacity, @Nullable Map<String, ?> configs) {
+ this.url = Objects.requireNonNull(url);
+ // TODO this does not seem to be used for Apicurio
+ this.identityMapCapacity = identityMapCapacity;
+ // Apicurio null pointers if this is left null.
Review Comment:
which property causes the NPE - identityMapCapacity or configs?
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCodergetSchemaTest.java:
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for schemacoder. */
+public class ApicurioSchemaRegistryCodergetSchemaTest {
+
+ @ParameterizedTest
+ @MethodSource("configProvider")
+ public void getSchemaId(TestSpec testSpec) {
+ try {
+
+ if (testSpec.extraConfigOptions == null) {
+ testSpec.extraConfigOptions = new HashMap<>();
+ testSpec.extraConfigOptions.put(USE_GLOBALID.key(), true);
+ testSpec.extraConfigOptions.put(USE_HEADERS.key(), true);
+ }
+
+ // get options map
+ Map<String, Object> registryConfigs =
+ getConfig(testSpec.isKey, testSpec.extraConfigOptions);
+ // this is just to have something in the stream
+ String str = readFile("src/test/resources/simple1.avro", StandardCharsets.UTF_8);
+
+ byte[] schemaBytes = str.getBytes(StandardCharsets.UTF_8);
+ InputStream in;
+ if (testSpec.byteArray == null) {
+ in = new ByteArrayInputStream(schemaBytes);
+ } else {
+ // the byteArray is the magic byte and schema id that should be at the start of the
+ // body
+ byte[] combined = new byte[testSpec.byteArray.length + schemaBytes.length];
+ ByteBuffer byteBuffer = ByteBuffer.wrap(combined);
+ byteBuffer.put(testSpec.byteArray);
+ byteBuffer.put(schemaBytes);
+ combined = byteBuffer.array();
+ in = new ByteArrayInputStream(combined);
+ }
+
+ ApicurioSchemaRegistryCoder apicurioSchemaRegistryCoder =
+ new ApicurioSchemaRegistryCoder(null, registryConfigs);
+ Map<String, Object> additionalProperties = new HashMap<>();
+ additionalProperties.put(apicurioSchemaRegistryCoder.HEADERS, testSpec.headers);
+ additionalProperties.put(apicurioSchemaRegistryCoder.IS_KEY, testSpec.isKey);
+ if (testSpec.extraConfigOptions == null) {
+ testSpec.extraConfigOptions = new HashMap<>();
+ }
+ if (testSpec.extraConfigOptions.get(USE_GLOBALID.key()) == null) {
+ testSpec.extraConfigOptions.put(USE_GLOBALID.key(), true);
+ }
+
+ boolean useGlobalId = (boolean) testSpec.extraConfigOptions.get(USE_GLOBALID.key());
+
+ Long testSchemaId =
+ apicurioSchemaRegistryCoder.getSchemaId(in, additionalProperties, useGlobalId);
+ assertThat(testSchemaId).isEqualTo(testSpec.expectedId);
+ assertThat(testSpec.expectSuccess).isTrue();
+ } catch (Exception e) {
+ if (testSpec.expectSuccess) {
+ throw new RuntimeException("Unexpected Error occurred", e);
+ }
+ }
+ }
+
+ private static class TestSpec {
+
+ boolean isKey;
+ Map<String, Object> headers;
+ byte[] byteArray;
+ Map<String, Object> extraConfigOptions;
+ long expectedId;
+ boolean expectSuccess;
+
+ private TestSpec(
+ boolean isKey,
+ Map<String, Object> headers,
+ byte[] byteArray,
+ Map<String, Object> extraConfigOptions,
+ long expectedId,
+ boolean expectSuccess) {
+ this.isKey = isKey;
+ this.headers = headers;
+ this.byteArray = byteArray;
+ this.extraConfigOptions = extraConfigOptions;
+ this.expectedId = expectedId;
+ this.expectSuccess = expectSuccess;
+ }
+
+ @Override
+ public String toString() {
+ return "TestSpec{"
+ + "isKey="
+ + isKey
+ + ", headers="
+ + headers
+ + ", byteArray="
+ + byteArray
+ + ", extraConfigOptions="
+ + extraConfigOptions
+ + ", expectedId="
+ + expectedId
+ + '}';
+ }
+ }
+
+ private static class ValidTestSpec extends TestSpec {
+ private ValidTestSpec(
+ boolean isKey,
+ Map<String, Object> headers,
+ byte[] byteArray,
+ Map<String, Object> extraConfigOptions,
+ long expectedId) {
+ super(isKey, headers, byteArray, extraConfigOptions, expectedId, true);
+ }
+ }
+
+ private static class InvalidTestSpec extends TestSpec {
+ private InvalidTestSpec(
+ boolean isKey,
+ Map<String, Object> headers,
+ byte[] byteArray,
+ Map<String, Object> extraConfigOptions,
+ long expectedId) {
+ super(isKey, headers, byteArray, extraConfigOptions, expectedId, false);
+ }
+ }
+
+ static Collection<TestSpec> configProvider() {
+ return ImmutableList.<TestSpec>builder()
+ .addAll(getValidTestSpecs())
+ .addAll(getInvalidTestSpecs())
+ .build();
+ }
+
+ @NotNull
+ private static ImmutableList<TestSpec> getValidTestSpecs() {
+ return ImmutableList.of(
+ // Key global
+ new ValidTestSpec(
+ true,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // Value global
+ new ValidTestSpec(
+ false,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // Key content
+ new ValidTestSpec(
+ true,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ ofEntries(
+ new String[] {AvroApicurioFormatOptions.USE_GLOBALID.key()},
+ new Object[] {false}),
+ 12L),
+ // Value content
+ new ValidTestSpec(
+ false,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ ofEntries(new String[] {USE_GLOBALID.key()}, new Object[] {false}),
+ 12L),
+ // test legacy global
+ new ValidTestSpec(
+ false,
+ null, // send no headers
+ getLegacyByteArray(12L),
+ ofEntries(new String[] {USE_HEADERS.key()}, new Object[] {false}),
+ 12L),
+ // test legacy content
+ new ValidTestSpec(
+ false,
+ null, // send no headers
+ getLegacyByteArray(12L),
+ ofEntries(
+ new String[] {USE_HEADERS.key(), USE_GLOBALID.key()},
+ new Object[] {false, false}),
+ 12L));
+ }
+
+ private static byte[] getLegacyByteArray(long schemaId) {
+ byte[] legacyByteArray = new byte[1 + Long.BYTES];
+ ByteBuffer byteBuffer = ByteBuffer.wrap(legacyByteArray);
+ // magic byte
+ byteBuffer.put(new byte[] {0});
+ // 8 byte schema id
+ byteBuffer.put(ApicurioSchemaRegistryCoder.longToBytes(schemaId));
+ return byteBuffer.array();
+ }
+
+ @NotNull
+ private static ImmutableList<TestSpec> getInvalidTestSpecs() {
+ return ImmutableList.of(
+
+ // ** tests for default configuration - global , headers
+
+ // Key, configured global, header key content
+ new InvalidTestSpec(
+ true,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // Key, configured global, header value content
+ new InvalidTestSpec(
+ true,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // Key, configured global, header value global
+ new InvalidTestSpec(
+ true,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // Value, configured global, header value content
+ new InvalidTestSpec(
+ false,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // Value, configured global, header key content
+ new InvalidTestSpec(
+ false,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // Value, configured global, header key global
+ new InvalidTestSpec(
+ false,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)}),
+ null,
+ null,
+ 12L),
+ // tests LEGACY with no magic byte
+ new InvalidTestSpec(
+ false,
+ null, // send no headers
+ null,
+ ofEntries(new String[] {USE_HEADERS.key()}, new Object[] {false}),
+ 12L));
+ }
+
+ // Cannot use Map.of or Map.ofEntries as we are Java 8.
+ private static Map<String, Object> ofEntries(String[] keys, Object[] values) {
+ HashMap map = new HashMap();
+ for (int i = 0; i < keys.length; i++) {
+ map.put(keys[i], values[i]);
+ }
+ return map;
+ }
+
+ @NotNull
+ private static Map<String, Object> getConfig(
+ boolean isKey, // TODO do we need to handle value and key specific map keys or not?
Review Comment:
// TODO
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/RegistryAvroFormatFactoryTest.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
+import org.apache.flink.formats.avro.RowDataToAvroConverters;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.TestDynamicTableFactory;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.avro.Schema;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the {@link ApicurioRegistryAvroFormatFactory}. */
+class RegistryAvroFormatFactoryTest {
+
+ private static final ResolvedSchema SCHEMA =
+ ResolvedSchema.of(
+ Column.physical("a", DataTypes.STRING()),
+ Column.physical("b", DataTypes.INT()),
+ Column.physical("c", DataTypes.BOOLEAN()));
+
+ private static final RowType ROW_TYPE =
+ (RowType) SCHEMA.toPhysicalRowDataType().getLogicalType();
+
+ private static final String REGISTRY_URL = "http://localhost:8080//apis/registry/v2";
+ private static final String SCHEMA_STRING =
+ "{\n"
+ + " \"type\": \"record\",\n"
+ + " \"name\": \"test_record\",\n"
+ + " \"fields\": [\n"
+ + " {\n"
+ + " \"name\": \"a\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"string\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " },\n"
+ + " {\n"
+ + " \"name\": \"b\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"int\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " },\n"
+ + " {\n"
+ + " \"name\": \"c\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"boolean\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " }\n"
+ + " ]\n"
+ + "}";
+
+ private static final Map<String, String> EXPECTED_OPTIONAL_PROPERTIES = new HashMap<>();
+
+ static {
+ EXPECTED_OPTIONAL_PROPERTIES.put(
+ "schema.registry.ssl.keystore.location", getAbsolutePath("/test-keystore.jks"));
+ EXPECTED_OPTIONAL_PROPERTIES.put("schema.registry.ssl.keystore.password", "123456");
+ EXPECTED_OPTIONAL_PROPERTIES.put(
+ "schema.registry.ssl.truststore.location", getAbsolutePath("/test-keystore.jks"));
+ EXPECTED_OPTIONAL_PROPERTIES.put("schema.registry.ssl.truststore.password", "123456");
+ EXPECTED_OPTIONAL_PROPERTIES.put("basic.auth.credentials.source", "USER_INFO");
+ EXPECTED_OPTIONAL_PROPERTIES.put("basic.auth.user.info", "user:pwd");
+ EXPECTED_OPTIONAL_PROPERTIES.put("bearer.auth.token", "CUSTOM");
+ }
+
+ /**
+ * Returns the full options modified by the given consumer {@code optionModifier}.
+ *
+ * @param optionModifier Consumer to modify the options
+ */
+ private Map<String, String> getModifiedOptions(Consumer<Map<String, String>> optionModifier) {
+ Map<String, String> options = getDefaultOptions();
+ optionModifier.accept(options);
+ return options;
+ }
+
+ @Test
+ void testbuildOptionalPropertiesMapEmpty() {
+ Set<ConfigOption<?>> configOptions = ApicurioRegistryAvroFormatFactory.getOptionalOptions();
+ Map<String, String> inputMap = new HashMap<>();
+ Configuration formatOptions = Configuration.fromMap(inputMap);
+ ApicurioRegistryAvroFormatFactory registryAvroFormatFactory =
+ new ApicurioRegistryAvroFormatFactory();
+ Map<String, Object> outputMap =
+ registryAvroFormatFactory.buildOptionalPropertiesMap(formatOptions);
+
+ for (ConfigOption configOption : configOptions) {
+ if (configOption.hasDefaultValue()) {
+ assertThat(outputMap.get(configOption.key()))
+ .isEqualTo(configOption.defaultValue());
+ } else {
+ assertThat(outputMap.get(configOption.key())).isNull();
+ }
+ }
+ // check some specific values to confirm
+ assertThat((boolean) outputMap.get(USE_GLOBALID.key())).isTrue();
+ assertThat(outputMap.get(SSL_TRUSTSTORE_LOCATION.key())).isNull();
+ }
+
+ @Test
+ void testbuildOptionalPropertiesMapWithConfig() {
+
+ Set<ConfigOption<?>> configOptions = ApicurioRegistryAvroFormatFactory.getOptionalOptions();
+ Map<String, String> inputMap = new HashMap<>();
+ for (ConfigOption configOption : configOptions) {
+ if (!configOption.equals(PROPERTIES)
+ && !configOption.equals(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)
+ && !configOption.key().equals(USE_GLOBALID)
+ && !configOption.key().equals(USE_HEADERS)) {
+
+ String value = configOption.key() + "-value";
+ inputMap.put(configOption.key(), value);
+ }
+ }
+
+ Configuration formatOptions = Configuration.fromMap(inputMap);
+ Duration durationTestValue = Duration.ofSeconds(10);
+ formatOptions.set(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION, durationTestValue);
+ formatOptions.set(USE_GLOBALID, true);
+ formatOptions.set(USE_HEADERS, true);
+ Map propertiesInput = new HashMap<>();
+ propertiesInput.put("aaa", "111");
+ propertiesInput.put("bbb", "222");
+ formatOptions.set(PROPERTIES, propertiesInput);
+ ApicurioRegistryAvroFormatFactory registryAvroFormatFactory =
+ new ApicurioRegistryAvroFormatFactory();
+ Map<String, Object> outputMap =
+ registryAvroFormatFactory.buildOptionalPropertiesMap(formatOptions);
+
+ for (ConfigOption configOption : configOptions) {
+ Object expectedValue;
+ if (configOption.equals(PROPERTIES)) {
+ expectedValue = propertiesInput;
+ } else if (configOption.equals(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)) {
+ expectedValue = durationTestValue;
+ } else if (configOption.equals(USE_HEADERS)) {
+ expectedValue = true;
+ } else if (configOption.equals(USE_GLOBALID)) {
+ expectedValue = true;
+ } else {
+ expectedValue = configOption.key() + "-value";
+ }
+ assertThat(outputMap.get(configOption.key())).isEqualTo(expectedValue);
+ }
+ }
+
+ 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", ApicurioRegistryAvroFormatFactory.IDENTIFIER);
+ options.put("avro-apicurio.url", REGISTRY_URL);
+ return options;
+ }
+
+ private Map<String, String> getOptionalProperties() {
+ final Map<String, String> properties = new HashMap<>();
+ // defined via Flink maintained options
+ properties.put(
+ AvroApicurioFormatOptions.SSL_KEYSTORE_LOCATION.key(),
+ getAbsolutePath("/test-keystore.jks"));
+ properties.put(AvroApicurioFormatOptions.SSL_KEYSTORE_PASSWORD.key(), "123456");
+ properties.put(
+ AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION.key(),
+ getAbsolutePath("/test-keystore.jks"));
+ properties.put(AvroApicurioFormatOptions.SSL_TRUSTSTORE_PASSWORD.key(), "123456");
+ properties.put(AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID.key(), "USERID");
+ properties.put(AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD.key(), "PASSWORD");
+ // defined via general property map
+ properties.put("schema", SCHEMA_STRING);
+
+ return getModifiedOptions(
+ opts ->
+ properties.forEach(
+ (k, v) ->
+ opts.put(
+ String.format(
+ "%s.%s",
+ ApicurioRegistryAvroFormatFactory
+ .IDENTIFIER,
+ k),
+ v)));
+ }
+
+ private static String getAbsolutePath(String path) {
+ try {
+ return CachedSchemaCoderProviderTest.class.getResource(path).toURI().getPath();
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Test
+ void testSerializationSchemaWithOptionalProperties() {
+ final AvroRowDataSerializationSchema expectedSer =
+ new AvroRowDataSerializationSchema(
+ ROW_TYPE,
+ ApicurioRegistryAvroSerializationSchema.forGeneric(
+ new Schema.Parser().parse(SCHEMA_STRING),
+ REGISTRY_URL,
+ EXPECTED_OPTIONAL_PROPERTIES),
+ RowDataToAvroConverters.createConverter(ROW_TYPE));
+
+ final DynamicTableSink actualSink = createTableSink(SCHEMA, getOptionalProperties());
+ assertThat(actualSink).isInstanceOf(TestDynamicTableFactory.DynamicTableSinkMock.class);
+ TestDynamicTableFactory.DynamicTableSinkMock sinkMock =
+ (TestDynamicTableFactory.DynamicTableSinkMock) actualSink;
+
+ SerializationSchema<RowData> actualSer =
+ sinkMock.valueFormat.createRuntimeEncoder(null, SCHEMA.toPhysicalRowDataType());
+
+ // assertThat(actualSer).isEqualTo(expectedSer);
Review Comment:
commented out code
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/RegistryAvroFormatFactoryTest.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
+import org.apache.flink.formats.avro.RowDataToAvroConverters;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.TestDynamicTableFactory;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.avro.Schema;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the {@link ApicurioRegistryAvroFormatFactory}. */
+class RegistryAvroFormatFactoryTest {
+
+ private static final ResolvedSchema SCHEMA =
+ ResolvedSchema.of(
+ Column.physical("a", DataTypes.STRING()),
+ Column.physical("b", DataTypes.INT()),
+ Column.physical("c", DataTypes.BOOLEAN()));
+
+ private static final RowType ROW_TYPE =
+ (RowType) SCHEMA.toPhysicalRowDataType().getLogicalType();
+
+ private static final String REGISTRY_URL = "http://localhost:8080//apis/registry/v2";
+ private static final String SCHEMA_STRING =
+ "{\n"
+ + " \"type\": \"record\",\n"
+ + " \"name\": \"test_record\",\n"
+ + " \"fields\": [\n"
+ + " {\n"
+ + " \"name\": \"a\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"string\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " },\n"
+ + " {\n"
+ + " \"name\": \"b\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"int\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " },\n"
+ + " {\n"
+ + " \"name\": \"c\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"boolean\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " }\n"
+ + " ]\n"
+ + "}";
+
+ private static final Map<String, String> EXPECTED_OPTIONAL_PROPERTIES = new HashMap<>();
+
+ static {
+ EXPECTED_OPTIONAL_PROPERTIES.put(
+ "schema.registry.ssl.keystore.location", getAbsolutePath("/test-keystore.jks"));
+ EXPECTED_OPTIONAL_PROPERTIES.put("schema.registry.ssl.keystore.password", "123456");
+ EXPECTED_OPTIONAL_PROPERTIES.put(
+ "schema.registry.ssl.truststore.location", getAbsolutePath("/test-keystore.jks"));
+ EXPECTED_OPTIONAL_PROPERTIES.put("schema.registry.ssl.truststore.password", "123456");
+ EXPECTED_OPTIONAL_PROPERTIES.put("basic.auth.credentials.source", "USER_INFO");
+ EXPECTED_OPTIONAL_PROPERTIES.put("basic.auth.user.info", "user:pwd");
+ EXPECTED_OPTIONAL_PROPERTIES.put("bearer.auth.token", "CUSTOM");
+ }
+
+ /**
+ * Returns the full options modified by the given consumer {@code optionModifier}.
+ *
+ * @param optionModifier Consumer to modify the options
+ */
+ private Map<String, String> getModifiedOptions(Consumer<Map<String, String>> optionModifier) {
+ Map<String, String> options = getDefaultOptions();
+ optionModifier.accept(options);
+ return options;
+ }
+
+ @Test
+ void testbuildOptionalPropertiesMapEmpty() {
+ Set<ConfigOption<?>> configOptions = ApicurioRegistryAvroFormatFactory.getOptionalOptions();
+ Map<String, String> inputMap = new HashMap<>();
+ Configuration formatOptions = Configuration.fromMap(inputMap);
+ ApicurioRegistryAvroFormatFactory registryAvroFormatFactory =
+ new ApicurioRegistryAvroFormatFactory();
+ Map<String, Object> outputMap =
+ registryAvroFormatFactory.buildOptionalPropertiesMap(formatOptions);
+
+ for (ConfigOption configOption : configOptions) {
+ if (configOption.hasDefaultValue()) {
+ assertThat(outputMap.get(configOption.key()))
+ .isEqualTo(configOption.defaultValue());
+ } else {
+ assertThat(outputMap.get(configOption.key())).isNull();
+ }
+ }
+ // check some specific values to confirm
+ assertThat((boolean) outputMap.get(USE_GLOBALID.key())).isTrue();
+ assertThat(outputMap.get(SSL_TRUSTSTORE_LOCATION.key())).isNull();
+ }
+
+ @Test
+ void testbuildOptionalPropertiesMapWithConfig() {
+
+ Set<ConfigOption<?>> configOptions = ApicurioRegistryAvroFormatFactory.getOptionalOptions();
+ Map<String, String> inputMap = new HashMap<>();
+ for (ConfigOption configOption : configOptions) {
+ if (!configOption.equals(PROPERTIES)
+ && !configOption.equals(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)
+ && !configOption.key().equals(USE_GLOBALID)
+ && !configOption.key().equals(USE_HEADERS)) {
+
+ String value = configOption.key() + "-value";
+ inputMap.put(configOption.key(), value);
+ }
+ }
+
+ Configuration formatOptions = Configuration.fromMap(inputMap);
+ Duration durationTestValue = Duration.ofSeconds(10);
+ formatOptions.set(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION, durationTestValue);
+ formatOptions.set(USE_GLOBALID, true);
+ formatOptions.set(USE_HEADERS, true);
+ Map propertiesInput = new HashMap<>();
+ propertiesInput.put("aaa", "111");
+ propertiesInput.put("bbb", "222");
+ formatOptions.set(PROPERTIES, propertiesInput);
+ ApicurioRegistryAvroFormatFactory registryAvroFormatFactory =
+ new ApicurioRegistryAvroFormatFactory();
+ Map<String, Object> outputMap =
+ registryAvroFormatFactory.buildOptionalPropertiesMap(formatOptions);
+
+ for (ConfigOption configOption : configOptions) {
+ Object expectedValue;
+ if (configOption.equals(PROPERTIES)) {
+ expectedValue = propertiesInput;
+ } else if (configOption.equals(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)) {
+ expectedValue = durationTestValue;
+ } else if (configOption.equals(USE_HEADERS)) {
+ expectedValue = true;
+ } else if (configOption.equals(USE_GLOBALID)) {
+ expectedValue = true;
+ } else {
+ expectedValue = configOption.key() + "-value";
+ }
+ assertThat(outputMap.get(configOption.key())).isEqualTo(expectedValue);
+ }
+ }
+
+ 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", ApicurioRegistryAvroFormatFactory.IDENTIFIER);
+ options.put("avro-apicurio.url", REGISTRY_URL);
+ return options;
+ }
+
+ private Map<String, String> getOptionalProperties() {
+ final Map<String, String> properties = new HashMap<>();
+ // defined via Flink maintained options
+ properties.put(
+ AvroApicurioFormatOptions.SSL_KEYSTORE_LOCATION.key(),
+ getAbsolutePath("/test-keystore.jks"));
+ properties.put(AvroApicurioFormatOptions.SSL_KEYSTORE_PASSWORD.key(), "123456");
+ properties.put(
+ AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION.key(),
+ getAbsolutePath("/test-keystore.jks"));
+ properties.put(AvroApicurioFormatOptions.SSL_TRUSTSTORE_PASSWORD.key(), "123456");
+ properties.put(AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID.key(), "USERID");
+ properties.put(AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD.key(), "PASSWORD");
+ // defined via general property map
+ properties.put("schema", SCHEMA_STRING);
+
+ return getModifiedOptions(
+ opts ->
+ properties.forEach(
+ (k, v) ->
+ opts.put(
+ String.format(
+ "%s.%s",
+ ApicurioRegistryAvroFormatFactory
+ .IDENTIFIER,
+ k),
+ v)));
+ }
+
+ private static String getAbsolutePath(String path) {
+ try {
+ return CachedSchemaCoderProviderTest.class.getResource(path).toURI().getPath();
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ @Test
+ void testSerializationSchemaWithOptionalProperties() {
+ final AvroRowDataSerializationSchema expectedSer =
+ new AvroRowDataSerializationSchema(
+ ROW_TYPE,
+ ApicurioRegistryAvroSerializationSchema.forGeneric(
+ new Schema.Parser().parse(SCHEMA_STRING),
+ REGISTRY_URL,
+ EXPECTED_OPTIONAL_PROPERTIES),
+ RowDataToAvroConverters.createConverter(ROW_TYPE));
+
+ final DynamicTableSink actualSink = createTableSink(SCHEMA, getOptionalProperties());
+ assertThat(actualSink).isInstanceOf(TestDynamicTableFactory.DynamicTableSinkMock.class);
+ TestDynamicTableFactory.DynamicTableSinkMock sinkMock =
+ (TestDynamicTableFactory.DynamicTableSinkMock) actualSink;
+
+ SerializationSchema<RowData> actualSer =
+ sinkMock.valueFormat.createRuntimeEncoder(null, SCHEMA.toPhysicalRowDataType());
+
+ // assertThat(actualSer).isEqualTo(expectedSer);
+ System.err.println("expectedSer:" + expectedSer + "\nactualSer:" + actualSer);
Review Comment:
System.err ?
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
Review Comment:
`// need flip` could be elaborated
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/AvroApicurioFormatOptions.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import io.apicurio.registry.serde.SerdeConfig;
+
+import java.time.Duration;
+import java.util.Map;
+
+/** Options for Schema Registry Avro format. */
+@PublicEvolving
+public class AvroApicurioFormatOptions {
+
+ public static final ConfigOption<String> URL =
+ ConfigOptions.key("url")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The URL of the Apicurio Registry to fetch/register schemas.");
+
+ public static final ConfigOption<Boolean> USE_HEADERS =
+ ConfigOptions.key("use-headers")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Used by serializers and deserializers. Configures the identifier for artifacts. \n"
+ + "true means use global ID, false means use content ID.\n"
+ + "Instructs the serializer to write the specified ID to Kafka,\n"
+ + "and instructs the deserializer to use this ID to find the schema.");
+
+ public static final ConfigOption<Boolean> USE_GLOBALID =
+ ConfigOptions.key("use-globalid")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Configures to read/write the artifact identifier to "
+ + "Kafka message headers instead of in the message payload.");
+
+ public static final ConfigOption<String> GROUP_ID =
+ ConfigOptions.key("groupId")
+ .stringType()
+ .defaultValue("default")
+ .withDescription("GroupId Used by deserializers.");
Review Comment:
s/Used/used/
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCodergetSchemaTest.java:
##########
@@ -0,0 +1,363 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for schemacoder. */
+public class ApicurioSchemaRegistryCodergetSchemaTest {
+
+ @ParameterizedTest
+ @MethodSource("configProvider")
+ public void getSchemaId(TestSpec testSpec) {
+ try {
+
+ if (testSpec.extraConfigOptions == null) {
+ testSpec.extraConfigOptions = new HashMap<>();
+ testSpec.extraConfigOptions.put(USE_GLOBALID.key(), true);
+ testSpec.extraConfigOptions.put(USE_HEADERS.key(), true);
+ }
+
+ // get options map
+ Map<String, Object> registryConfigs =
+ getConfig(testSpec.isKey, testSpec.extraConfigOptions);
+ // this is just to have something in the stream
+ String str = readFile("src/test/resources/simple1.avro", StandardCharsets.UTF_8);
+
+ byte[] schemaBytes = str.getBytes(StandardCharsets.UTF_8);
+ InputStream in;
+ if (testSpec.byteArray == null) {
+ in = new ByteArrayInputStream(schemaBytes);
+ } else {
+ // the byteArray is the magic byte and schema id that should be at the start of the
+ // body
+ byte[] combined = new byte[testSpec.byteArray.length + schemaBytes.length];
+ ByteBuffer byteBuffer = ByteBuffer.wrap(combined);
+ byteBuffer.put(testSpec.byteArray);
+ byteBuffer.put(schemaBytes);
+ combined = byteBuffer.array();
+ in = new ByteArrayInputStream(combined);
+ }
+
+ ApicurioSchemaRegistryCoder apicurioSchemaRegistryCoder =
+ new ApicurioSchemaRegistryCoder(null, registryConfigs);
+ Map<String, Object> additionalProperties = new HashMap<>();
+ additionalProperties.put(apicurioSchemaRegistryCoder.HEADERS, testSpec.headers);
+ additionalProperties.put(apicurioSchemaRegistryCoder.IS_KEY, testSpec.isKey);
+ if (testSpec.extraConfigOptions == null) {
Review Comment:
This repeats lines 54-55 above
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/AvroApicurioFormatOptions.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import io.apicurio.registry.serde.SerdeConfig;
+
+import java.time.Duration;
+import java.util.Map;
+
+/** Options for Schema Registry Avro format. */
+@PublicEvolving
+public class AvroApicurioFormatOptions {
+
+ public static final ConfigOption<String> URL =
+ ConfigOptions.key("url")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The URL of the Apicurio Registry to fetch/register schemas.");
+
+ public static final ConfigOption<Boolean> USE_HEADERS =
+ ConfigOptions.key("use-headers")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Used by serializers and deserializers. Configures the identifier for artifacts. \n"
+ + "true means use global ID, false means use content ID.\n"
+ + "Instructs the serializer to write the specified ID to Kafka,\n"
+ + "and instructs the deserializer to use this ID to find the schema.");
+
+ public static final ConfigOption<Boolean> USE_GLOBALID =
+ ConfigOptions.key("use-globalid")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Configures to read/write the artifact identifier to "
Review Comment:
"Configures to read/write the artifact id" could be reworded
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
+ // so we need to get the artifact references, find the latest version,
+ // using its groupId and artifactId we can get the globalId, from which we can get
+ // a de-referenced schema
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " got using contentId " + schemaId);
+ }
+ List<ArtifactReference> artifactReferenceList =
+ registryClient.getArtifactReferencesByContentId(schemaId);
+ if (artifactReferenceList == null || artifactReferenceList.isEmpty()) {
+ throw new IOException(
+ "Could not find any schemas in the Apicurio Registry with content ID "
+ + schemaId);
+ }
+ long latestVersion = 0L;
+ String artifactId = "";
+ String groupId = "";
+ for (ArtifactReference artifactReference : artifactReferenceList) {
+ long version = Long.valueOf(artifactReference.getVersion());
+ // TODO check Apicurio code to see how it checks for the latest version
+ if (version > latestVersion) {
+ latestVersion = version;
+ artifactId = artifactReference.getArtifactId();
+ groupId = artifactReference.getGroupId();
+ }
+ }
+ VersionMetaData artifactVersion =
+ registryClient.getArtifactVersionMetaData(
+ groupId, artifactId, latestVersion + "");
+ schemaId = artifactVersion.getGlobalId();
+ }
+ // get the schema in canonical form with references dereferenced
+ InputStream schemaInputStream = registryClient.getContentByGlobalId(schemaId, true, true);
+ Schema schema = new Schema.Parser().parse(schemaInputStream);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ methodName
+ + " got schema "
+ + schema
+ + " using "
+ + (useGlobalId ? "global" : "content")
+ + " ID "
+ + schemaId);
+ }
+ return schema;
+ }
+
+ /**
+ * Get the schema id.
+ *
+ * @param in the Kafka body as an input stream
+ * @param additionalParameters additional Pararamters containing the headers and whetehr we are
+ * processing a key or value
+ * @param useGlobalID hen true globalIDs are used, otherwise contentIDS
+ * @return the id of the schema
+ * @throws IOException error occurred
+ */
+ protected long getSchemaId(
+ InputStream in, Map<String, Object> additionalParameters, boolean useGlobalID)
+ throws IOException {
+ long schemaId = 0L;
+ Map<String, Object> headers = (Map<String, Object>) additionalParameters.get(HEADERS);
+ boolean isKey = (boolean) additionalParameters.get(IS_KEY);
+ boolean useHeaders = (boolean) configs.get(USE_HEADERS.key());
+
+ if (useHeaders) {
+ String globalIDHeaderName =
+ isKey ? APICURIO_KEY_GLOBAL_ID_HEADER : APICURIO_VALUE_GLOBAL_ID_HEADER;
+ String contentIDHeaderName =
+ isKey ? APICURIO_KEY_CONTENT_ID_HEADER : APICURIO_VALUE_CONTENT_ID_HEADER;
+ // get from headers
+ if (headers.get(globalIDHeaderName) != null) {
+ if (!useGlobalID) {
+ throw new IOException(
+ "The message has a global ID header "
+ + globalIDHeaderName
+ + ", but the configuration is for content ID.\n"
+ + "Header names received: \n"
+ + headers.keySet().stream().collect(Collectors.joining(",")));
+ }
+ byte[] globalIDByteArray = (byte[]) headers.get(globalIDHeaderName);
+ schemaId = bytesToLong(globalIDByteArray);
+ } else if (headers.get(contentIDHeaderName) != null) {
+ if (useGlobalID) {
+ throw new IOException(
+ "The message has a content ID header "
+ + contentIDHeaderName
+ + "\n"
+ + ", but the configuration is for global ID.\n"
+ + "Header names received: "
+ + headers.keySet().stream().collect(Collectors.joining(",")));
+ }
+ byte[] contentIDByteArray = (byte[]) headers.get(contentIDHeaderName);
+ schemaId = bytesToLong(contentIDByteArray);
+ } else {
+ String headerName = useGlobalID ? globalIDHeaderName : contentIDHeaderName;
+ throw new IOException(
+ "The format was configured to enable headers, but the expected Kafka header "
+ + headerName
+ + " is not present.\n "
+ + "Header names received: "
+ + headers.keySet().stream().collect(Collectors.joining(","))
+ + "\n "
+ + "Ensure that the kafka message was created by an Apicurio client. Check that it is "
+ + "sending the appropriate ID in the header, if the message is sending the ID in the payload,\n"
+ + " review and amend the format configuration options "
+ + USE_HEADERS
+ + " and "
+ + USE_GLOBALID);
+ }
+ } else {
+ // the id is in the payload
+ DataInputStream dataInputStream = new DataInputStream(in);
+ if (dataInputStream.readByte() != 0) {
+ throw new IOException(
+ "Unknown data format. Magic number was not found. "
+ + USE_HEADERS
+ + "= false"
+ + " configs="
+ + configs);
+ }
+ schemaId = dataInputStream.readLong();
+ }
+ return schemaId;
+ }
+
+ @Override
+ public void writeSchema(Schema schema, OutputStream out) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector is being used, upgrade the Kafka connector.");
+ }
+
+ /**
+ * The Avro schema will be written depending on the configuration.
+ *
+ * @param schema Avro Schema to write into the output stream,
+ * @param out the output stream where the schema is serialized.
+ * @param inputProperties properties containing the topic name
+ * @param outputProperties headers to populate
+ * @throws IOException Exception has occurred
+ */
+ @Override
+ public void writeSchema(
+ Schema schema,
+ OutputStream out,
+ Map<String, Object> inputProperties,
+ Map<String, Object> outputProperties)
+ throws IOException {
+ String methodName = "writeSchema with additional properties";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+
+ boolean useHeaders = (boolean) configs.get(USE_HEADERS.key());
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+ boolean isKey = (boolean) inputProperties.get(IS_KEY);
+
+ String groupId = (String) configs.get(AvroApicurioFormatOptions.GROUP_ID.key());
+ Object artifactId = configs.get(AvroApicurioFormatOptions.REGISTERED_ARTIFACT_ID.key());
+ Object artifactName = configs.get(AvroApicurioFormatOptions.REGISTERED_ARTIFACT_NAME.key());
+ if (inputProperties.get(TOPIC_NAME) == null) {
+ String configOptionName = null;
+ if (artifactId == null) {
+ configOptionName = "artifactId";
+ }
+ if (artifactName == null) {
+ configOptionName = "artifactName";
+ }
+ if (configOptionName != null) {
+ throw new IOException(
+ "No topic name was configured, and it is required to provide a default for "
+ + configOptionName);
+ }
+ }
+ String defaultArtifactName =
+ (String) inputProperties.get(TOPIC_NAME + (isKey ? "-key" : "-value"));
+ artifactId = (artifactId == null) ? defaultArtifactName : artifactId;
+ artifactName = (artifactName == null) ? defaultArtifactName : artifactName;
+
+ // config option has a default so no need to check for null
+ String artifactDescription =
+ (String)
Review Comment:
formatting
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoderwriteSchemaTest.java:
##########
@@ -0,0 +1,478 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import org.apache.avro.Schema;
+import org.apache.commons.collections.MapUtils;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for schemacoder. */
+public class ApicurioSchemaRegistryCoderwriteSchemaTest {
+
+ @ParameterizedTest
+ @MethodSource("configProvider")
+ public void writeSchema(TestSpec testSpec) throws IOException {
+ try {
+ if (testSpec.avroFileName == null) {
+ testSpec.avroFileName = "src/test/resources/simpleschema1.avsc";
+ }
+
+ // get schema from passed fileName contents
+ String schemaStr = readFile(testSpec.avroFileName, StandardCharsets.UTF_8);
+ Schema schema = new Schema.Parser().parse(schemaStr);
+ // get config
+ Map<String, Object> configs = getConfig(testSpec.isKey, testSpec.extraConfigOptions);
+ // setup input
+ Map<String, Object> inputProperties = new HashMap<>();
+ inputProperties.put(ApicurioSchemaRegistryCoder.TOPIC_NAME, testSpec.topicName);
+ inputProperties.put(ApicurioSchemaRegistryCoder.IS_KEY, testSpec.isKey);
+ // setup output
+ OutputStream out = new ByteArrayOutputStream();
+ Map<String, Object> outputProperties = new HashMap<>();
+
+ // create Mock registryClient
+ MockRegistryClient registryClient = new MockRegistryClient();
+ ArtifactMetaData artifactMetaData = new ArtifactMetaData();
+ artifactMetaData.setContentId(testSpec.contentId);
+ artifactMetaData.setGlobalId(testSpec.globalId);
+ registryClient.setCreateArtifactResult(artifactMetaData);
+
+ ApicurioSchemaRegistryCoder apicurioSchemaRegistryCoder =
+ new ApicurioSchemaRegistryCoder(registryClient, configs);
+
+ apicurioSchemaRegistryCoder.writeSchema(schema, out, inputProperties, outputProperties);
+ Map<String, Object> headersFromWriteSchema =
+ (Map<String, Object>) outputProperties.get(ApicurioSchemaRegistryCoder.HEADERS);
+
+ if (testSpec.extraConfigOptions == null) {
+ testSpec.extraConfigOptions = new HashMap<>();
+ }
+ if (testSpec.extraConfigOptions.get(USE_GLOBALID.key()) == null) {
+ testSpec.extraConfigOptions.put(USE_GLOBALID.key(), true);
+ }
+
+ if (testSpec.extraConfigOptions.get(USE_HEADERS.key()) == null) {
+ testSpec.extraConfigOptions.put(USE_HEADERS.key(), true);
+ }
+
+ boolean useGlobalId = (boolean) testSpec.extraConfigOptions.get(USE_GLOBALID.key());
+ boolean useHeaders = (boolean) testSpec.extraConfigOptions.get(USE_HEADERS.key());
+
+ Long testProducedId;
+
+ if (useHeaders) {
+ String expectedHeaderName;
+ if (useGlobalId) {
+ if (testSpec.isKey) {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER;
+ } else {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER;
+ }
+ } else {
+ if (testSpec.isKey) {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER;
+ } else {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_CONTENT_ID_HEADER;
+ }
+ }
+ byte[] byteArrayFromHeaders =
+ (byte[]) headersFromWriteSchema.get(expectedHeaderName);
+ assertThat(byteArrayFromHeaders).isNotEmpty();
+ testProducedId = apicurioSchemaRegistryCoder.bytesToLong(byteArrayFromHeaders);
+ byte[] byteArrayExpectedHeader =
+ (byte[]) testSpec.expectedHeaders.get(expectedHeaderName);
+ if (!testSpec.expectSuccess && byteArrayExpectedHeader == null) {
+ // this is an error test so fail with an exception
+ throw new RuntimeException(
+ "Error test found an unexpected header with testSpec" + testSpec);
+ }
+ assertThat(byteArrayExpectedHeader).isNotEmpty();
+ long expectedHeaderId =
+ apicurioSchemaRegistryCoder.bytesToLong(byteArrayExpectedHeader);
+ if (!testSpec.expectSuccess) {
+ throw new RuntimeException(
+ "Error test: id in header is not expected with testSpec" + testSpec);
+ }
+ assertThat(testProducedId).isEqualTo(expectedHeaderId);
+ } else {
+ ByteArrayOutputStream baos = (ByteArrayOutputStream) out;
+ byte[] bytes = baos.toByteArray();
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ DataInputStream dataInputStream = new DataInputStream(bais);
+ // check for the magic byte
+ assertThat(dataInputStream.readByte()).isEqualTo((byte) 0);
+ testProducedId = dataInputStream.readLong();
+ }
+ long expectedId;
+ if (useGlobalId) {
+ expectedId = testSpec.globalId;
+ } else {
+ expectedId = testSpec.contentId;
+ }
+ assertThat(testProducedId).isEqualTo(expectedId);
+ assertThat(testSpec.expectSuccess).isTrue();
+ } catch (Exception e) {
+ if (testSpec.expectSuccess) {
+ throw new RuntimeException("Unexpected Error occurred with testSpec" + testSpec, e);
+ }
+ }
+ }
+
+ private static class TestSpec {
+
+ boolean isKey;
+ Map<String, Object> extraConfigOptions;
+ String topicName;
+ String avroFileName;
+ long contentId;
+ long globalId;
+ Map<String, Object> expectedHeaders;
+ boolean expectSuccess;
+
+ private TestSpec(
+ boolean isKey,
+ Map<String, Object> extraConfigOptions,
+ String topicName,
+ String avroFileName,
+ long contentId,
+ long globalId,
+ Map<String, Object> expectedHeaders,
+ boolean expectSuccess) {
+ this.extraConfigOptions = extraConfigOptions;
+ this.isKey = isKey;
+ this.topicName = topicName;
+ this.avroFileName = avroFileName;
+ this.contentId = contentId;
+ this.globalId = globalId;
+ this.expectedHeaders = expectedHeaders;
+ this.expectSuccess = expectSuccess;
+ }
+
+ @Override
+ public String toString() {
+ String extraConfigOptionsStr = "";
+ if (MapUtils.isNotEmpty(this.extraConfigOptions)) {
+ for (String key : this.extraConfigOptions.keySet()) {
+ extraConfigOptionsStr =
+ extraConfigOptionsStr
+ + "("
+ + key
+ + ","
+ + this.extraConfigOptions.get(key)
+ + ")\n";
+ }
+ }
+ String headersStr = "";
+ if (MapUtils.isNotEmpty(this.expectedHeaders)) {
+ for (String key : this.expectedHeaders.keySet()) {
+ headersStr =
+ headersStr + "(" + key + "," + this.expectedHeaders.get(key) + ")\n";
+ }
+ }
+ return "TestSpec{"
+ + "extraConfigOptions="
+ + extraConfigOptionsStr
+ + ", headers="
+ + headersStr
+ + ", isKey="
+ + isKey
+ + ", topicName="
+ + topicName
+ + ", avroFileName="
+ + avroFileName
+ + ", contentId="
+ + contentId
+ + ", globalId="
+ + globalId
+ + ", expectedHeaders="
+ + expectedHeaders
+ + ", expectSuccess="
+ + expectSuccess
+ + '}';
+ }
+ }
+
+ private static class ValidTestSpec extends TestSpec {
+ private ValidTestSpec(
+ boolean isKey,
+ Map<String, Object> extraConfigOptions,
+ String topicName,
+ String avroFileName,
+ long contentId,
+ long globalId,
+ Map<String, Object> expectedHeaders) {
+ super(
+ isKey,
+ extraConfigOptions,
+ topicName,
+ avroFileName,
+ contentId,
+ globalId,
+ expectedHeaders,
+ true);
+ }
+ }
+
+ private static class InvalidTestSpec extends TestSpec {
+ private InvalidTestSpec(
+ boolean isKey,
+ Map<String, Object> extraConfigOptions,
+ String topicName,
+ String avroFileName,
+ long contentId,
+ long globalId,
+ Map<String, Object> expectedHeaders) {
+ super(
+ isKey,
+ extraConfigOptions,
+ topicName,
+ avroFileName,
+ contentId,
+ globalId,
+ expectedHeaders,
+ false);
+ }
+ }
+
+ static Collection<TestSpec> configProvider() {
+ return ImmutableList.<TestSpec>builder()
+ .addAll(getValidTestSpecs())
+ .addAll(getInvalidTestSpecs())
+ .build();
+ }
+
+ @NotNull
+ private static ImmutableList<TestSpec> getValidTestSpecs() {
+ return ImmutableList.of(
+ // Key global
+ new ValidTestSpec(
+ true,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13)})),
+ // value global
+ new ValidTestSpec(
+ false,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13)})),
+ // key content
+ new ValidTestSpec(
+ true,
+ ofEntries(new String[] {USE_GLOBALID.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)})),
+ // value content
+ new ValidTestSpec(
+ false,
+ ofEntries(new String[] {USE_GLOBALID.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)})),
+ // Key global Legacy
+ new ValidTestSpec(
+ true,
+ ofEntries(new String[] {USE_HEADERS.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null),
+ // Value global Legacy
+ new ValidTestSpec(
+ false,
+ ofEntries(new String[] {USE_HEADERS.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null),
+ // Key content Legacy
+ new ValidTestSpec(
+ true,
+ ofEntries(
+ new String[] {USE_HEADERS.key(), USE_GLOBALID.key()},
+ new Object[] {false, false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null),
+ // Value content Legacy
+ new ValidTestSpec(
+ false,
+ ofEntries(
+ new String[] {USE_HEADERS.key(), USE_GLOBALID.key()},
+ new Object[] {false, false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null));
+ }
+
+ private static byte[] getLegacyByteArray(long schemaId) {
+ byte[] legacyByteArray = new byte[1 + Long.BYTES];
+ ByteBuffer byteBuffer = ByteBuffer.wrap(legacyByteArray);
+ // magic byte
+ byteBuffer.put(new byte[] {0});
+ // 8 byte schema id
+ byteBuffer.put(ApicurioSchemaRegistryCoder.longToBytes(schemaId));
+ return byteBuffer.array();
+ }
+
+ @NotNull
+ private static ImmutableList<TestSpec> getInvalidTestSpecs() {
+ return ImmutableList.of(
+ // global key with value header
+ new InvalidTestSpec(
+ true,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13L)})),
+ // global key with mismatching content id in global header
+ new InvalidTestSpec(
+ true,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)})),
+ // global key with mismatching global id in content header
+ new InvalidTestSpec(
+ true,
+ ofEntries(new String[] {USE_GLOBALID.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13)}))
+ // Note policing Legacy mode with a header (when the id should be in
+ // the payload)
+ );
+ }
+
+ // Cannot use Map.of or Map.ofEntries as we are Java 8.
+ private static Map<String, Object> ofEntries(String[] keys, Object[] values) {
+ HashMap map = new HashMap();
Review Comment:
`Map<String, Object> map = new HashMap<>();`
##########
flink-formats/flink-avro-apicurio-registry/src/main/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoder.java:
##########
@@ -0,0 +1,379 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.formats.avro.AvroFormatOptions;
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import io.apicurio.registry.rest.client.RegistryClient;
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import io.apicurio.registry.rest.v2.beans.ArtifactReference;
+import io.apicurio.registry.rest.v2.beans.IfExists;
+import io.apicurio.registry.rest.v2.beans.VersionMetaData;
+import io.apicurio.registry.types.ArtifactType;
+import org.apache.avro.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+
+/** Reads and Writes schema using Avro Schema Registry protocol. */
+public class ApicurioSchemaRegistryCoder implements SchemaCoder {
+
+ public static final String APICURIO_KEY_GLOBAL_ID_HEADER = "apicurio.key.globalId";
+
+ public static final String APICURIO_KEY_CONTENT_ID_HEADER = "apicurio.key.contentId";
+
+ public static final String APICURIO_VALUE_GLOBAL_ID_HEADER = "apicurio.value.globalId";
+
+ public static final String APICURIO_VALUE_CONTENT_ID_HEADER = "apicurio.value.contentId";
+
+ public static final String APICURIO_VALUE_ENCODING = "apicurio.value.encoding";
+
+ private final RegistryClient registryClient;
+
+ private final Map<String, Object> configs;
+
+ private static final int MAGIC_BYTE = 0;
+
+ public static final String TOPIC_NAME = "TOPIC_NAME";
+
+ public static final String HEADERS = "HEADERS";
+
+ public static final String IS_KEY = "IS_KEY";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ApicurioSchemaRegistryCoder.class);
+
+ /**
+ * Creates {@link SchemaCoder} that uses provided {@link RegistryClient} to connect to schema
+ * registry.
+ *
+ * @param registryClient client to connect schema registry
+ * @param configs map for registry configs
+ */
+ public ApicurioSchemaRegistryCoder(RegistryClient registryClient, Map<String, Object> configs) {
+ this.registryClient = registryClient;
+ this.configs = configs;
+ }
+
+ @Override
+ public Schema readSchema(InputStream in) throws IOException {
+ throw new IOException(
+ "Incompatible Kafka connector. Use a Kafka connector that passes headers");
+ }
+
+ public long bytesToLong(byte[] bytes) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.put(bytes);
+ buffer.flip(); // need flip
+ return buffer.getLong();
+ }
+
+ public static byte[] longToBytes(long x) {
+ ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
+ buffer.putLong(x);
+ return buffer.array();
+ }
+
+ /**
+ * Get the Avro schema using the Apicurio Registry. In order to call the registry, we need to
+ * get the globalId of the Avro Schema in the registry. If the format is configured to expect
+ * headers, then the globalId will be obtained from the headers map. Otherwise, the globalId is
+ * obtained from the message payload, depending on the format configuration. This format is only
+ * used with the Kafka connector. The Kafka connector lives in another repository. This
+ * repository does not have any Kafka dependencies; it is therefore not possible for us to use
+ * the Kafka consumer record here passing a deserialization handler. So we have to get the
+ * information we need directly from the headers or the payload itself.
+ *
+ * @param in input stream
+ * @param additionalParameters map of additional parameters - including the headers an whether
+ * this is a key or value
+ * @return the Avro Schema
+ * @throws IOException if there is an error
+ */
+ @Override
+ public Schema readSchemaWithAdditionalParameters(
+ InputStream in, Map<String, Object> additionalParameters) throws IOException {
+ String methodName = "readSchemaWithAdditionalParameters";
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(methodName + " entered");
+ }
+ if (in == null) {
+ return null;
+ }
+ boolean useGlobalId = (boolean) configs.get(USE_GLOBALID.key());
+
+ long schemaId = getSchemaId(in, additionalParameters, useGlobalId);
+
+ if (!useGlobalId) {
+ // if we are using the content ID there is not get that will dereference.
Review Comment:
typo in this comment block, 'there is not get'
##########
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDataSerializationSchema.java:
##########
@@ -114,12 +115,33 @@ public void open(InitializationContext context) throws Exception {
@Override
public byte[] serialize(RowData row) {
+ GenericRecord record = null;
try {
// convert to record
- final GenericRecord record = (GenericRecord) runtimeConverter.convert(schema, row);
+ record = (GenericRecord) runtimeConverter.convert(schema, row);
return nestedSchema.serialize(record);
} catch (Exception e) {
- throw new RuntimeException("Failed to serialize row.", e);
+ throw new RuntimeException("Failed to serialize row1. record=" + record, e);
Review Comment:
row1 ?
##########
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/RegistryAvroSerializationSchema.java:
##########
@@ -107,17 +108,37 @@ public static RegistryAvroSerializationSchema<GenericRecord> forGeneric(
}
@Override
- public byte[] serialize(T object) {
- checkAvroInitialized();
+ public byte[] serialize(T element) {
+ return serializeWithAdditionalProperties(element, null, null);
+ }
+ @Override
+ public byte[] serializeWithAdditionalProperties(
+ T object,
+ Map<String, Object> inputAdditionalProperties,
+ Map<String, Object> outputAdditionalProperties) {
+ checkAvroInitialized();
if (object == null) {
return null;
} else {
try {
ByteArrayOutputStream outputStream = getOutputStream();
outputStream.reset();
Encoder encoder = getEncoder();
- schemaCoder.writeSchema(getSchema(), outputStream);
+ if (inputAdditionalProperties == null) {
+ throw new RuntimeException(
+ "inputAdditionalProperties"
+ + inputAdditionalProperties
+ + ",out"
+ + outputAdditionalProperties);
+ // schemaCoder.writeSchema(getSchema(), outputStream);
Review Comment:
commented out code
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/RegistryAvroFormatFactoryTest.java:
##########
@@ -0,0 +1,261 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
+import org.apache.flink.formats.avro.RowDataToAvroConverters;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.TestDynamicTableFactory;
+import org.apache.flink.table.types.logical.RowType;
+
+import org.apache.avro.Schema;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.PROPERTIES;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the {@link ApicurioRegistryAvroFormatFactory}. */
+class RegistryAvroFormatFactoryTest {
+
+ private static final ResolvedSchema SCHEMA =
+ ResolvedSchema.of(
+ Column.physical("a", DataTypes.STRING()),
+ Column.physical("b", DataTypes.INT()),
+ Column.physical("c", DataTypes.BOOLEAN()));
+
+ private static final RowType ROW_TYPE =
+ (RowType) SCHEMA.toPhysicalRowDataType().getLogicalType();
+
+ private static final String REGISTRY_URL = "http://localhost:8080//apis/registry/v2";
+ private static final String SCHEMA_STRING =
+ "{\n"
+ + " \"type\": \"record\",\n"
+ + " \"name\": \"test_record\",\n"
+ + " \"fields\": [\n"
+ + " {\n"
+ + " \"name\": \"a\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"string\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " },\n"
+ + " {\n"
+ + " \"name\": \"b\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"int\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " },\n"
+ + " {\n"
+ + " \"name\": \"c\",\n"
+ + " \"type\": [\n"
+ + " \"null\",\n"
+ + " \"boolean\"\n"
+ + " ],\n"
+ + " \"default\": null\n"
+ + " }\n"
+ + " ]\n"
+ + "}";
+
+ private static final Map<String, String> EXPECTED_OPTIONAL_PROPERTIES = new HashMap<>();
+
+ static {
+ EXPECTED_OPTIONAL_PROPERTIES.put(
+ "schema.registry.ssl.keystore.location", getAbsolutePath("/test-keystore.jks"));
+ EXPECTED_OPTIONAL_PROPERTIES.put("schema.registry.ssl.keystore.password", "123456");
+ EXPECTED_OPTIONAL_PROPERTIES.put(
+ "schema.registry.ssl.truststore.location", getAbsolutePath("/test-keystore.jks"));
+ EXPECTED_OPTIONAL_PROPERTIES.put("schema.registry.ssl.truststore.password", "123456");
+ EXPECTED_OPTIONAL_PROPERTIES.put("basic.auth.credentials.source", "USER_INFO");
+ EXPECTED_OPTIONAL_PROPERTIES.put("basic.auth.user.info", "user:pwd");
+ EXPECTED_OPTIONAL_PROPERTIES.put("bearer.auth.token", "CUSTOM");
+ }
+
+ /**
+ * Returns the full options modified by the given consumer {@code optionModifier}.
+ *
+ * @param optionModifier Consumer to modify the options
+ */
+ private Map<String, String> getModifiedOptions(Consumer<Map<String, String>> optionModifier) {
+ Map<String, String> options = getDefaultOptions();
+ optionModifier.accept(options);
+ return options;
+ }
+
+ @Test
+ void testbuildOptionalPropertiesMapEmpty() {
+ Set<ConfigOption<?>> configOptions = ApicurioRegistryAvroFormatFactory.getOptionalOptions();
+ Map<String, String> inputMap = new HashMap<>();
+ Configuration formatOptions = Configuration.fromMap(inputMap);
+ ApicurioRegistryAvroFormatFactory registryAvroFormatFactory =
+ new ApicurioRegistryAvroFormatFactory();
+ Map<String, Object> outputMap =
+ registryAvroFormatFactory.buildOptionalPropertiesMap(formatOptions);
+
+ for (ConfigOption configOption : configOptions) {
+ if (configOption.hasDefaultValue()) {
+ assertThat(outputMap.get(configOption.key()))
+ .isEqualTo(configOption.defaultValue());
+ } else {
+ assertThat(outputMap.get(configOption.key())).isNull();
+ }
+ }
+ // check some specific values to confirm
+ assertThat((boolean) outputMap.get(USE_GLOBALID.key())).isTrue();
+ assertThat(outputMap.get(SSL_TRUSTSTORE_LOCATION.key())).isNull();
+ }
+
+ @Test
+ void testbuildOptionalPropertiesMapWithConfig() {
+
+ Set<ConfigOption<?>> configOptions = ApicurioRegistryAvroFormatFactory.getOptionalOptions();
+ Map<String, String> inputMap = new HashMap<>();
+ for (ConfigOption configOption : configOptions) {
+ if (!configOption.equals(PROPERTIES)
+ && !configOption.equals(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)
+ && !configOption.key().equals(USE_GLOBALID)
+ && !configOption.key().equals(USE_HEADERS)) {
+
+ String value = configOption.key() + "-value";
+ inputMap.put(configOption.key(), value);
+ }
+ }
+
+ Configuration formatOptions = Configuration.fromMap(inputMap);
+ Duration durationTestValue = Duration.ofSeconds(10);
+ formatOptions.set(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION, durationTestValue);
+ formatOptions.set(USE_GLOBALID, true);
+ formatOptions.set(USE_HEADERS, true);
+ Map propertiesInput = new HashMap<>();
+ propertiesInput.put("aaa", "111");
+ propertiesInput.put("bbb", "222");
+ formatOptions.set(PROPERTIES, propertiesInput);
+ ApicurioRegistryAvroFormatFactory registryAvroFormatFactory =
+ new ApicurioRegistryAvroFormatFactory();
+ Map<String, Object> outputMap =
+ registryAvroFormatFactory.buildOptionalPropertiesMap(formatOptions);
+
+ for (ConfigOption configOption : configOptions) {
+ Object expectedValue;
+ if (configOption.equals(PROPERTIES)) {
+ expectedValue = propertiesInput;
+ } else if (configOption.equals(OIDC_AUTH_TOKEN_EXPIRATION_REDUCTION)) {
+ expectedValue = durationTestValue;
+ } else if (configOption.equals(USE_HEADERS)) {
+ expectedValue = true;
+ } else if (configOption.equals(USE_GLOBALID)) {
+ expectedValue = true;
+ } else {
+ expectedValue = configOption.key() + "-value";
+ }
+ assertThat(outputMap.get(configOption.key())).isEqualTo(expectedValue);
+ }
+ }
+
+ 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", ApicurioRegistryAvroFormatFactory.IDENTIFIER);
+ options.put("avro-apicurio.url", REGISTRY_URL);
+ return options;
+ }
+
+ private Map<String, String> getOptionalProperties() {
+ final Map<String, String> properties = new HashMap<>();
+ // defined via Flink maintained options
+ properties.put(
+ AvroApicurioFormatOptions.SSL_KEYSTORE_LOCATION.key(),
+ getAbsolutePath("/test-keystore.jks"));
+ properties.put(AvroApicurioFormatOptions.SSL_KEYSTORE_PASSWORD.key(), "123456");
+ properties.put(
+ AvroApicurioFormatOptions.SSL_TRUSTSTORE_LOCATION.key(),
+ getAbsolutePath("/test-keystore.jks"));
+ properties.put(AvroApicurioFormatOptions.SSL_TRUSTSTORE_PASSWORD.key(), "123456");
+ properties.put(AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_USERID.key(), "USERID");
+ properties.put(AvroApicurioFormatOptions.BASIC_AUTH_CREDENTIALS_PASSWORD.key(), "PASSWORD");
+ // defined via general property map
+ properties.put("schema", SCHEMA_STRING);
+
+ return getModifiedOptions(
Review Comment:
Could the formatting of this block be condensed?
##########
flink-formats/flink-avro-apicurio-registry/src/test/java/org/apache/flink/formats/avro/registry/apicurio/ApicurioSchemaRegistryCoderwriteSchemaTest.java:
##########
@@ -0,0 +1,478 @@
+/*
+ * 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.apicurio;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+
+import io.apicurio.registry.rest.v2.beans.ArtifactMetaData;
+import org.apache.avro.Schema;
+import org.apache.commons.collections.MapUtils;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_GLOBALID;
+import static org.apache.flink.formats.avro.registry.apicurio.AvroApicurioFormatOptions.USE_HEADERS;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for schemacoder. */
+public class ApicurioSchemaRegistryCoderwriteSchemaTest {
+
+ @ParameterizedTest
+ @MethodSource("configProvider")
+ public void writeSchema(TestSpec testSpec) throws IOException {
+ try {
+ if (testSpec.avroFileName == null) {
+ testSpec.avroFileName = "src/test/resources/simpleschema1.avsc";
+ }
+
+ // get schema from passed fileName contents
+ String schemaStr = readFile(testSpec.avroFileName, StandardCharsets.UTF_8);
+ Schema schema = new Schema.Parser().parse(schemaStr);
+ // get config
+ Map<String, Object> configs = getConfig(testSpec.isKey, testSpec.extraConfigOptions);
+ // setup input
+ Map<String, Object> inputProperties = new HashMap<>();
+ inputProperties.put(ApicurioSchemaRegistryCoder.TOPIC_NAME, testSpec.topicName);
+ inputProperties.put(ApicurioSchemaRegistryCoder.IS_KEY, testSpec.isKey);
+ // setup output
+ OutputStream out = new ByteArrayOutputStream();
+ Map<String, Object> outputProperties = new HashMap<>();
+
+ // create Mock registryClient
+ MockRegistryClient registryClient = new MockRegistryClient();
+ ArtifactMetaData artifactMetaData = new ArtifactMetaData();
+ artifactMetaData.setContentId(testSpec.contentId);
+ artifactMetaData.setGlobalId(testSpec.globalId);
+ registryClient.setCreateArtifactResult(artifactMetaData);
+
+ ApicurioSchemaRegistryCoder apicurioSchemaRegistryCoder =
+ new ApicurioSchemaRegistryCoder(registryClient, configs);
+
+ apicurioSchemaRegistryCoder.writeSchema(schema, out, inputProperties, outputProperties);
+ Map<String, Object> headersFromWriteSchema =
+ (Map<String, Object>) outputProperties.get(ApicurioSchemaRegistryCoder.HEADERS);
+
+ if (testSpec.extraConfigOptions == null) {
+ testSpec.extraConfigOptions = new HashMap<>();
+ }
+ if (testSpec.extraConfigOptions.get(USE_GLOBALID.key()) == null) {
+ testSpec.extraConfigOptions.put(USE_GLOBALID.key(), true);
+ }
+
+ if (testSpec.extraConfigOptions.get(USE_HEADERS.key()) == null) {
+ testSpec.extraConfigOptions.put(USE_HEADERS.key(), true);
+ }
+
+ boolean useGlobalId = (boolean) testSpec.extraConfigOptions.get(USE_GLOBALID.key());
+ boolean useHeaders = (boolean) testSpec.extraConfigOptions.get(USE_HEADERS.key());
+
+ Long testProducedId;
+
+ if (useHeaders) {
+ String expectedHeaderName;
+ if (useGlobalId) {
+ if (testSpec.isKey) {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER;
+ } else {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER;
+ }
+ } else {
+ if (testSpec.isKey) {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER;
+ } else {
+ expectedHeaderName =
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_CONTENT_ID_HEADER;
+ }
+ }
+ byte[] byteArrayFromHeaders =
+ (byte[]) headersFromWriteSchema.get(expectedHeaderName);
+ assertThat(byteArrayFromHeaders).isNotEmpty();
+ testProducedId = apicurioSchemaRegistryCoder.bytesToLong(byteArrayFromHeaders);
+ byte[] byteArrayExpectedHeader =
+ (byte[]) testSpec.expectedHeaders.get(expectedHeaderName);
+ if (!testSpec.expectSuccess && byteArrayExpectedHeader == null) {
+ // this is an error test so fail with an exception
+ throw new RuntimeException(
+ "Error test found an unexpected header with testSpec" + testSpec);
+ }
+ assertThat(byteArrayExpectedHeader).isNotEmpty();
+ long expectedHeaderId =
+ apicurioSchemaRegistryCoder.bytesToLong(byteArrayExpectedHeader);
+ if (!testSpec.expectSuccess) {
+ throw new RuntimeException(
+ "Error test: id in header is not expected with testSpec" + testSpec);
+ }
+ assertThat(testProducedId).isEqualTo(expectedHeaderId);
+ } else {
+ ByteArrayOutputStream baos = (ByteArrayOutputStream) out;
+ byte[] bytes = baos.toByteArray();
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ DataInputStream dataInputStream = new DataInputStream(bais);
+ // check for the magic byte
+ assertThat(dataInputStream.readByte()).isEqualTo((byte) 0);
+ testProducedId = dataInputStream.readLong();
+ }
+ long expectedId;
+ if (useGlobalId) {
+ expectedId = testSpec.globalId;
+ } else {
+ expectedId = testSpec.contentId;
+ }
+ assertThat(testProducedId).isEqualTo(expectedId);
+ assertThat(testSpec.expectSuccess).isTrue();
+ } catch (Exception e) {
+ if (testSpec.expectSuccess) {
+ throw new RuntimeException("Unexpected Error occurred with testSpec" + testSpec, e);
+ }
+ }
+ }
+
+ private static class TestSpec {
+
+ boolean isKey;
+ Map<String, Object> extraConfigOptions;
+ String topicName;
+ String avroFileName;
+ long contentId;
+ long globalId;
+ Map<String, Object> expectedHeaders;
+ boolean expectSuccess;
+
+ private TestSpec(
+ boolean isKey,
+ Map<String, Object> extraConfigOptions,
+ String topicName,
+ String avroFileName,
+ long contentId,
+ long globalId,
+ Map<String, Object> expectedHeaders,
+ boolean expectSuccess) {
+ this.extraConfigOptions = extraConfigOptions;
+ this.isKey = isKey;
+ this.topicName = topicName;
+ this.avroFileName = avroFileName;
+ this.contentId = contentId;
+ this.globalId = globalId;
+ this.expectedHeaders = expectedHeaders;
+ this.expectSuccess = expectSuccess;
+ }
+
+ @Override
+ public String toString() {
+ String extraConfigOptionsStr = "";
+ if (MapUtils.isNotEmpty(this.extraConfigOptions)) {
+ for (String key : this.extraConfigOptions.keySet()) {
+ extraConfigOptionsStr =
+ extraConfigOptionsStr
+ + "("
+ + key
+ + ","
+ + this.extraConfigOptions.get(key)
+ + ")\n";
+ }
+ }
+ String headersStr = "";
+ if (MapUtils.isNotEmpty(this.expectedHeaders)) {
+ for (String key : this.expectedHeaders.keySet()) {
+ headersStr =
+ headersStr + "(" + key + "," + this.expectedHeaders.get(key) + ")\n";
+ }
+ }
+ return "TestSpec{"
+ + "extraConfigOptions="
+ + extraConfigOptionsStr
+ + ", headers="
+ + headersStr
+ + ", isKey="
+ + isKey
+ + ", topicName="
+ + topicName
+ + ", avroFileName="
+ + avroFileName
+ + ", contentId="
+ + contentId
+ + ", globalId="
+ + globalId
+ + ", expectedHeaders="
+ + expectedHeaders
+ + ", expectSuccess="
+ + expectSuccess
+ + '}';
+ }
+ }
+
+ private static class ValidTestSpec extends TestSpec {
+ private ValidTestSpec(
+ boolean isKey,
+ Map<String, Object> extraConfigOptions,
+ String topicName,
+ String avroFileName,
+ long contentId,
+ long globalId,
+ Map<String, Object> expectedHeaders) {
+ super(
+ isKey,
+ extraConfigOptions,
+ topicName,
+ avroFileName,
+ contentId,
+ globalId,
+ expectedHeaders,
+ true);
+ }
+ }
+
+ private static class InvalidTestSpec extends TestSpec {
+ private InvalidTestSpec(
+ boolean isKey,
+ Map<String, Object> extraConfigOptions,
+ String topicName,
+ String avroFileName,
+ long contentId,
+ long globalId,
+ Map<String, Object> expectedHeaders) {
+ super(
+ isKey,
+ extraConfigOptions,
+ topicName,
+ avroFileName,
+ contentId,
+ globalId,
+ expectedHeaders,
+ false);
+ }
+ }
+
+ static Collection<TestSpec> configProvider() {
+ return ImmutableList.<TestSpec>builder()
+ .addAll(getValidTestSpecs())
+ .addAll(getInvalidTestSpecs())
+ .build();
+ }
+
+ @NotNull
+ private static ImmutableList<TestSpec> getValidTestSpecs() {
+ return ImmutableList.of(
+ // Key global
+ new ValidTestSpec(
+ true,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13)})),
+ // value global
+ new ValidTestSpec(
+ false,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13)})),
+ // key content
+ new ValidTestSpec(
+ true,
+ ofEntries(new String[] {USE_GLOBALID.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)})),
+ // value content
+ new ValidTestSpec(
+ false,
+ ofEntries(new String[] {USE_GLOBALID.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)})),
+ // Key global Legacy
+ new ValidTestSpec(
+ true,
+ ofEntries(new String[] {USE_HEADERS.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null),
+ // Value global Legacy
+ new ValidTestSpec(
+ false,
+ ofEntries(new String[] {USE_HEADERS.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null),
+ // Key content Legacy
+ new ValidTestSpec(
+ true,
+ ofEntries(
+ new String[] {USE_HEADERS.key(), USE_GLOBALID.key()},
+ new Object[] {false, false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null),
+ // Value content Legacy
+ new ValidTestSpec(
+ false,
+ ofEntries(
+ new String[] {USE_HEADERS.key(), USE_GLOBALID.key()},
+ new Object[] {false, false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ null));
+ }
+
+ private static byte[] getLegacyByteArray(long schemaId) {
+ byte[] legacyByteArray = new byte[1 + Long.BYTES];
+ ByteBuffer byteBuffer = ByteBuffer.wrap(legacyByteArray);
+ // magic byte
+ byteBuffer.put(new byte[] {0});
+ // 8 byte schema id
+ byteBuffer.put(ApicurioSchemaRegistryCoder.longToBytes(schemaId));
+ return byteBuffer.array();
+ }
+
+ @NotNull
+ private static ImmutableList<TestSpec> getInvalidTestSpecs() {
+ return ImmutableList.of(
+ // global key with value header
+ new InvalidTestSpec(
+ true,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_VALUE_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13L)})),
+ // global key with mismatching content id in global header
+ new InvalidTestSpec(
+ true,
+ null,
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_GLOBAL_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(12)})),
+ // global key with mismatching global id in content header
+ new InvalidTestSpec(
+ true,
+ ofEntries(new String[] {USE_GLOBALID.key()}, new Object[] {false}),
+ "topic1",
+ null,
+ 12L,
+ 13L,
+ ofEntries(
+ new String[] {
+ ApicurioSchemaRegistryCoder.APICURIO_KEY_CONTENT_ID_HEADER
+ },
+ new Object[] {ApicurioSchemaRegistryCoder.longToBytes(13)}))
+ // Note policing Legacy mode with a header (when the id should be in
+ // the payload)
+ );
+ }
+
+ // Cannot use Map.of or Map.ofEntries as we are Java 8.
+ private static Map<String, Object> ofEntries(String[] keys, Object[] values) {
+ HashMap map = new HashMap();
+ for (int i = 0; i < keys.length; i++) {
+ map.put(keys[i], values[i]);
+ }
+ return map;
+ }
+
+ @NotNull
+ private static Map<String, Object> getConfig(
+ boolean isKey, // TODO do we need to handle value and key specific map keys or not?
Review Comment:
// TODO
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org