You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/02/21 08:54:16 UTC
[flink] branch release-1.17 updated: [FLINK-30948][Formats/AWS] Remove GlueSchemaRegistry Avro and JSON formats from main repository
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.17 by this push:
new d5882933555 [FLINK-30948][Formats/AWS] Remove GlueSchemaRegistry Avro and JSON formats from main repository
d5882933555 is described below
commit d58829335557dac6ce428df5a80d4244fccf4491
Author: Hong Liang Teoh <li...@amazon.com>
AuthorDate: Sun Feb 19 10:34:28 2023 +0000
[FLINK-30948][Formats/AWS] Remove GlueSchemaRegistry Avro and JSON formats from main repository
---
azure-pipelines.yml | 2 -
.../flink-glue-schema-registry-avro-test/pom.xml | 245 ------------
.../registry/test/GSRKinesisPubsubClient.java | 113 ------
.../test/GlueSchemaRegistryAvroKinesisITCase.java | 237 -----------
.../src/test/resources/avro/user.avsc | 27 --
.../src/test/resources/log4j2-test.properties | 28 --
.../flink-glue-schema-registry-json-test/pom.xml | 229 -----------
.../registry/test/json/GSRKinesisPubsubClient.java | 107 -----
.../json/GlueSchemaRegistryJsonKinesisITCase.java | 211 ----------
.../src/test/resources/log4j2-test.properties | 28 --
flink-end-to-end-tests/pom.xml | 2 -
.../3e77c07d-dfdb-4ec0-8e64-5fad5c651c72 | 0
.../dc78e80c-3bb3-45bb-87c1-b57472d6f45b | 0
.../archunit-violations/stored.rules | 4 -
.../flink-avro-glue-schema-registry/pom.xml | 144 -------
...lueSchemaRegistryAvroDeserializationSchema.java | 90 -----
.../GlueSchemaRegistryAvroSchemaCoder.java | 86 ----
.../GlueSchemaRegistryAvroSchemaCoderProvider.java | 53 ---
.../GlueSchemaRegistryAvroSerializationSchema.java | 133 -------
.../GlueSchemaRegistryInputStreamDeserializer.java | 90 -----
.../GlueSchemaRegistryOutputStreamSerializer.java | 97 -----
.../architecture/TestCodeArchitectureTest.java | 40 --
...chemaRegistryAvroDeserializationSchemaTest.java | 64 ---
.../GlueSchemaRegistryAvroSchemaCoderTest.java | 221 -----------
...eSchemaRegistryAvroSerializationSchemaTest.java | 172 --------
...eSchemaRegistryInputStreamDeserializerTest.java | 311 ---------------
...ueSchemaRegistryOutputStreamSerializerTest.java | 133 -------
.../formats/avro/glue/schema/registry/User.java | 434 ---------------------
.../src/test/java/resources/avro/user.avsc | 27 --
.../org.junit.jupiter.api.extension.Extension | 16 -
.../src/test/resources/archunit.properties | 31 --
.../4703059b-4f06-41c9-9724-644e6d00584f | 0
.../b99819a4-a946-475e-883f-963de77c7e57 | 0
.../archunit-violations/stored.rules | 4 -
.../flink-json-glue-schema-registry/pom.xml | 156 --------
...lueSchemaRegistryJsonDeserializationSchema.java | 95 -----
.../GlueSchemaRegistryJsonSchemaCoder.java | 119 ------
.../GlueSchemaRegistryJsonSchemaCoderProvider.java | 51 ---
.../GlueSchemaRegistryJsonSerializationSchema.java | 77 ----
.../architecture/TestCodeArchitectureTest.java | 40 --
.../formats/json/glue/schema/registry/Car.java | 89 -----
...chemaRegistryJsonDeserializationSchemaTest.java | 182 ---------
.../GlueSchemaRegistryJsonSchemaCoderTest.java | 80 ----
...eSchemaRegistryJsonSerializationSchemaTest.java | 224 -----------
.../org.junit.jupiter.api.extension.Extension | 16 -
.../src/test/resources/archunit.properties | 31 --
flink-formats/pom.xml | 2 -
tools/azure-pipelines/build-apache-repo.yml | 2 -
tools/azure-pipelines/e2e-template.yml | 2 -
tools/azure-pipelines/jobs-template.yml | 2 -
tools/ci/stage.sh | 2 -
tools/maven/suppressions.xml | 3 -
52 files changed, 4552 deletions(-)
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index 38f755ef222..76dbdc07bf2 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -60,8 +60,6 @@ variables:
SECRET_S3_BUCKET: $[variables.IT_CASE_S3_BUCKET]
SECRET_S3_ACCESS_KEY: $[variables.IT_CASE_S3_ACCESS_KEY]
SECRET_S3_SECRET_KEY: $[variables.IT_CASE_S3_SECRET_KEY]
- SECRET_GLUE_SCHEMA_ACCESS_KEY: $[variables.IT_CASE_GLUE_SCHEMA_ACCESS_KEY]
- SECRET_GLUE_SCHEMA_SECRET_KEY: $[variables.IT_CASE_GLUE_SCHEMA_SECRET_KEY]
stages:
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/pom.xml b/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/pom.xml
deleted file mode 100644
index b218aa81ce7..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/pom.xml
+++ /dev/null
@@ -1,245 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <parent>
- <artifactId>flink-end-to-end-tests</artifactId>
- <groupId>org.apache.flink</groupId>
- <version>1.17-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
-
- <artifactId>flink-glue-schema-registry-avro-test_${scala.binary.version}</artifactId>
- <name>Flink : E2E Tests : Avro AWS Glue Schema Registry</name>
- <packaging>jar</packaging>
-
- <properties>
- <aws.sdk.version>1.12.276</aws.sdk.version>
- <aws.sdkv2.version>2.19.14</aws.sdkv2.version>
- <kotlin.version>1.7.10</kotlin.version>
- </properties>
-
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-annotations</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-metrics-core</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-base</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-base</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>bom</artifactId>
- <version>${aws.sdkv2.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>aws-java-sdk-bom</artifactId>
- <version>${aws.sdk.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
-
- <!-- For dependency convergence -->
- <dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- <version>3.21.7</version>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>30.0-jre</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-reflect</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-common</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-jdk8</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-scripting-compiler-embeddable</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlinx</groupId>
- <artifactId>kotlinx-serialization-core-jvm</artifactId>
- <version>1.0.1</version>
- </dependency>
- <dependency>
- <groupId>org.checkerframework</groupId>
- <artifactId>checker-qual</artifactId>
- <version>3.5.0</version>
- </dependency>
- <dependency>
- <groupId>com.squareup.okio</groupId>
- <artifactId>okio</artifactId>
- <version>2.8.0</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains</groupId>
- <artifactId>annotations</artifactId>
- <version>17.0.0</version>
- </dependency>
- <dependency>
- <groupId>com.github.docker-java</groupId>
- <artifactId>docker-java-api</artifactId>
- <version>3.2.13</version>
- </dependency>
- <dependency>
- <groupId>com.github.docker-java</groupId>
- <artifactId>docker-java-transport-zerodep</artifactId>
- <version>3.2.13</version>
- </dependency>
- <dependency>
- <groupId>com.github.docker-java</groupId>
- <artifactId>docker-java-transport</artifactId>
- <version>3.2.13</version>
- </dependency>
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>amazon-kinesis-client</artifactId>
- <version>1.14.7</version>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-clients</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <!-- Temporary: The tests need to be refactored to remove these dependencies -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kinesis</artifactId>
- <version>1.16.0</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kinesis</artifactId>
- <version>1.16.0</version>
- <type>test-jar</type>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-kinesis-streams</artifactId>
- <version>1.16.0</version>
- <type>test-jar</type>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-avro</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-avro-glue-schema-registry_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-base</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-test-utils-junit</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-guava</artifactId>
- </dependency>
-
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <systemPropertyVariables>
- <!-- Required for Kinesalite. -->
- <!-- Including shaded and non-shaded conf to support test running from Maven and IntelliJ -->
- <com.amazonaws.sdk.disableCbor>true</com.amazonaws.sdk.disableCbor>
- <com.amazonaws.sdk.disableCertChecking>true</com.amazonaws.sdk.disableCertChecking>
- <org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCbor>true</org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCbor>
- <org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCertChecking>true</org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCertChecking>
- </systemPropertyVariables>
- </configuration>
- </plugin>
- </plugins>
- </build>
-
-</project>
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/java/org/apache/flink/glue/schema/registry/test/GSRKinesisPubsubClient.java b/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/java/org/apache/flink/glue/schema/registry/test/GSRKinesisPubsubClient.java
deleted file mode 100644
index 0559845285c..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/java/org/apache/flink/glue/schema/registry/test/GSRKinesisPubsubClient.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * 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.glue.schema.registry.test;
-
-import org.apache.flink.streaming.connectors.kinesis.testutils.KinesisPubsubClient;
-
-import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
-import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import com.amazonaws.services.schemaregistry.utils.AvroRecordType;
-import org.apache.avro.generic.GenericRecord;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-
-/**
- * Simple client to publish and retrieve messages, using the AWS Kinesis SDK, Flink Kinesis
- * Connectors and Glue Schema Registry classes.
- */
-public class GSRKinesisPubsubClient {
- private final KinesisPubsubClient client;
- private final GlueSchemaRegistrySerializationFacade serializationFacade;
- private final GlueSchemaRegistryDeserializationFacade deserializationFacade;
-
- public GSRKinesisPubsubClient(
- Properties properties, AwsCredentialsProvider gsrCredentialsProvider) {
- this.client = new KinesisPubsubClient(properties);
- this.serializationFacade = createSerializationFacade(gsrCredentialsProvider);
- this.deserializationFacade = createDeserializationFacade(gsrCredentialsProvider);
- }
-
- public void sendMessage(String schema, String streamName, GenericRecord msg) {
- UUID schemaVersionId =
- serializationFacade.getOrRegisterSchemaVersion(
- AWSSerializerInput.builder()
- .schemaDefinition(schema)
- .dataFormat(DataFormat.AVRO.name())
- .schemaName(streamName)
- .transportName(streamName)
- .build());
-
- client.sendMessage(
- streamName, serializationFacade.serialize(DataFormat.AVRO, msg, schemaVersionId));
- }
-
- public List<Object> readAllMessages(String streamName) throws Exception {
-
- return client.readAllMessages(
- streamName,
- bytes ->
- deserializationFacade.deserialize(
- AWSDeserializerInput.builder()
- .buffer(ByteBuffer.wrap(bytes))
- .transportName(streamName)
- .build()));
- }
-
- public void createStream(String stream, int shards, Properties props) throws Exception {
- client.createTopic(stream, shards, props);
- }
-
- private Map<String, Object> getSerDeConfigs() {
- Map<String, Object> configs = new HashMap<>();
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "ca-central-1");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
- configs.put(
- AWSSchemaRegistryConstants.AVRO_RECORD_TYPE,
- AvroRecordType.GENERIC_RECORD.getName());
-
- return configs;
- }
-
- private GlueSchemaRegistrySerializationFacade createSerializationFacade(
- final AwsCredentialsProvider credentialsProvider) {
- return GlueSchemaRegistrySerializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .glueSchemaRegistryConfiguration(
- new GlueSchemaRegistryConfiguration(getSerDeConfigs()))
- .build();
- }
-
- private GlueSchemaRegistryDeserializationFacade createDeserializationFacade(
- final AwsCredentialsProvider credentialsProvider) {
- return GlueSchemaRegistryDeserializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .configs(getSerDeConfigs())
- .build();
- }
-}
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/java/org/apache/flink/glue/schema/registry/test/GlueSchemaRegistryAvroKinesisITCase.java b/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/java/org/apache/flink/glue/schema/registry/test/GlueSchemaRegistryAvroKinesisITCase.java
deleted file mode 100644
index 3a41abc2cd7..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/java/org/apache/flink/glue/schema/registry/test/GlueSchemaRegistryAvroKinesisITCase.java
+++ /dev/null
@@ -1,237 +0,0 @@
-/*
- * 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.glue.schema.registry.test;
-
-import org.apache.flink.api.common.time.Deadline;
-import org.apache.flink.connectors.kinesis.testutils.KinesaliteContainer;
-import org.apache.flink.formats.avro.glue.schema.registry.GlueSchemaRegistryAvroDeserializationSchema;
-import org.apache.flink.formats.avro.glue.schema.registry.GlueSchemaRegistryAvroSerializationSchema;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer;
-import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer;
-import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants;
-import org.apache.flink.util.StringUtils;
-import org.apache.flink.util.TestLogger;
-
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import com.amazonaws.services.schemaregistry.utils.AvroRecordType;
-import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericData;
-import org.apache.avro.generic.GenericRecord;
-import org.junit.After;
-import org.junit.Assume;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.Timeout;
-import org.testcontainers.containers.Network;
-import org.testcontainers.utility.DockerImageName;
-import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
-import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
-import software.amazon.awssdk.core.SdkSystemSetting;
-
-import java.io.IOException;
-import java.net.URL;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_ACCESS_KEY_ID;
-import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_SECRET_ACCESS_KEY;
-import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.STREAM_INITIAL_POSITION;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** End-to-end test for Glue Schema Registry AVRO format using Kinesalite. */
-public class GlueSchemaRegistryAvroKinesisITCase extends TestLogger {
- private static final String INPUT_STREAM = "gsr_avro_input_stream";
- private static final String OUTPUT_STREAM = "gsr_avro_output_stream";
- private static final String INTER_CONTAINER_KINESALITE_ALIAS = "kinesalite";
- private static final String ACCESS_KEY = System.getenv("IT_CASE_GLUE_SCHEMA_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("IT_CASE_GLUE_SCHEMA_SECRET_KEY");
-
- private static final Network network = Network.newNetwork();
-
- @ClassRule public static final Timeout TIMEOUT = new Timeout(10, TimeUnit.MINUTES);
-
- @ClassRule
- public static final KinesaliteContainer KINESALITE =
- new KinesaliteContainer(
- DockerImageName.parse("instructure/kinesalite").withTag("latest"))
- .withNetwork(network)
- .withNetworkAliases(INTER_CONTAINER_KINESALITE_ALIAS);
-
- private GSRKinesisPubsubClient kinesisClient;
-
- @Before
- public void setUp() throws Exception {
- Assume.assumeTrue(
- "Access key not configured, skipping test...",
- !StringUtils.isNullOrWhitespaceOnly(ACCESS_KEY));
- Assume.assumeTrue(
- "Secret key not configured, skipping test...",
- !StringUtils.isNullOrWhitespaceOnly(SECRET_KEY));
-
- StaticCredentialsProvider gsrCredentialsProvider =
- StaticCredentialsProvider.create(
- AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY));
-
- Properties properties = KINESALITE.getContainerProperties();
-
- kinesisClient = new GSRKinesisPubsubClient(properties, gsrCredentialsProvider);
- kinesisClient.createStream(INPUT_STREAM, 2, properties);
- kinesisClient.createStream(OUTPUT_STREAM, 2, properties);
-
- System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");
- }
-
- @After
- public void teardown() {
- System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());
- }
-
- @Test
- public void testGSRAvroGenericFormatWithFlink() throws Exception {
- List<GenericRecord> messages = getRecords();
- for (GenericRecord msg : messages) {
- kinesisClient.sendMessage(getSchema().toString(), INPUT_STREAM, msg);
- }
- log.info("generated records");
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
-
- DataStream<GenericRecord> input = env.addSource(createSource());
- input.addSink(createSink());
- env.executeAsync();
-
- Deadline deadline = Deadline.fromNow(Duration.ofSeconds(60));
- List<Object> results = kinesisClient.readAllMessages(OUTPUT_STREAM);
- while (deadline.hasTimeLeft() && results.size() < messages.size()) {
- log.info("waiting for results..");
- Thread.sleep(1000);
- results = kinesisClient.readAllMessages(OUTPUT_STREAM);
- }
- log.info("results: {}", results);
-
- assertThat(results).containsExactlyInAnyOrderElementsOf(messages);
- }
-
- private FlinkKinesisConsumer<GenericRecord> createSource() throws Exception {
- Properties properties = KINESALITE.getContainerProperties();
- properties.setProperty(
- STREAM_INITIAL_POSITION,
- ConsumerConfigConstants.InitialPosition.TRIM_HORIZON.name());
- return new FlinkKinesisConsumer<>(
- INPUT_STREAM,
- GlueSchemaRegistryAvroDeserializationSchema.forGeneric(getSchema(), getConfigs()),
- properties);
- }
-
- private FlinkKinesisProducer<GenericRecord> createSink() throws Exception {
- FlinkKinesisProducer<GenericRecord> producer =
- new FlinkKinesisProducer<>(
- GlueSchemaRegistryAvroSerializationSchema.forGeneric(
- getSchema(), OUTPUT_STREAM, getConfigs()),
- getProducerProperties());
- producer.setDefaultStream(OUTPUT_STREAM);
- producer.setDefaultPartition("fakePartition");
- return producer;
- }
-
- private Properties getProducerProperties() throws Exception {
- Properties producerProperties = KINESALITE.getContainerProperties();
- // producer needs region even when URL is specified
- producerProperties.put(ConsumerConfigConstants.AWS_REGION, "ca-central-1");
- // test driver does not deaggregate
- producerProperties.put("AggregationEnabled", String.valueOf(false));
-
- // KPL does not recognize endpoint URL..
- String kinesisUrl = producerProperties.getProperty(ConsumerConfigConstants.AWS_ENDPOINT);
- if (kinesisUrl != null) {
- URL url = new URL(kinesisUrl);
- producerProperties.put("KinesisEndpoint", url.getHost());
- producerProperties.put("KinesisPort", Integer.toString(url.getPort()));
- producerProperties.put("VerifyCertificate", "false");
- }
- return producerProperties;
- }
-
- private Schema getSchema() throws IOException {
- Schema.Parser parser = new Schema.Parser();
-
- return parser.parse(
- GlueSchemaRegistryAvroKinesisITCase.class
- .getClassLoader()
- .getResourceAsStream("avro/user.avsc"));
- }
-
- private Map<String, Object> getConfigs() {
- Map<String, Object> configs = new HashMap<>();
- configs.put(AWS_ACCESS_KEY_ID, ACCESS_KEY);
- configs.put(AWS_SECRET_ACCESS_KEY, SECRET_KEY);
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "ca-central-1");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
- configs.put(
- AWSSchemaRegistryConstants.AVRO_RECORD_TYPE,
- AvroRecordType.GENERIC_RECORD.getName());
-
- return configs;
- }
-
- private List<GenericRecord> getRecords() throws IOException {
- Schema userSchema = getSchema();
-
- GenericRecord sansa = new GenericData.Record(userSchema);
- sansa.put("name", "Sansa");
- sansa.put("favorite_number", 99);
- sansa.put("favorite_color", "white");
-
- GenericRecord harry = new GenericData.Record(userSchema);
- harry.put("name", "Harry");
- harry.put("favorite_number", 10);
- harry.put("favorite_color", "black");
-
- GenericRecord hermione = new GenericData.Record(userSchema);
- hermione.put("name", "Hermione");
- hermione.put("favorite_number", 1);
- hermione.put("favorite_color", "red");
-
- GenericRecord ron = new GenericData.Record(userSchema);
- ron.put("name", "Ron");
- ron.put("favorite_number", 18);
- ron.put("favorite_color", "green");
-
- GenericRecord jay = new GenericData.Record(userSchema);
- jay.put("name", "Jay");
- jay.put("favorite_number", 0);
- jay.put("favorite_color", "blue");
-
- List<GenericRecord> records = new ArrayList<>();
- records.add(sansa);
- records.add(harry);
- records.add(hermione);
- records.add(ron);
- records.add(jay);
- return records;
- }
-}
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/resources/avro/user.avsc b/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/resources/avro/user.avsc
deleted file mode 100644
index 343b5d3e35e..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/resources/avro/user.avsc
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-{
- "namespace": "org.apache.flink.glue.schema.registry.test",
- "type": "record",
- "name": "User",
- "fields": [
- {"name": "name", "type": "string" },
- {"name": "favorite_number", "type": ["int", "null"] },
- {"name": "favorite_color", "type": ["string", "null"] }
- ]
-}
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/resources/log4j2-test.properties b/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/resources/log4j2-test.properties
deleted file mode 100644
index e463a0e1a08..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-avro-test/src/test/resources/log4j2-test.properties
+++ /dev/null
@@ -1,28 +0,0 @@
-################################################################################
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-################################################################################
-
-# Set root logger level to OFF to not flood build logs
-# set manually to INFO for debugging purposes
-rootLogger.level = INFO
-rootLogger.appenderRef.test.ref = TestLogger
-
-appender.testlogger.name = TestLogger
-appender.testlogger.type = CONSOLE
-appender.testlogger.target = SYSTEM_ERR
-appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/pom.xml b/flink-end-to-end-tests/flink-glue-schema-registry-json-test/pom.xml
deleted file mode 100644
index b1582c4e15f..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/pom.xml
+++ /dev/null
@@ -1,229 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <parent>
- <artifactId>flink-end-to-end-tests</artifactId>
- <groupId>org.apache.flink</groupId>
- <version>1.17-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
-
- <artifactId>flink-glue-schema-registry-json-test</artifactId>
- <name>Flink : E2E Tests : JSON AWS Glue Schema Registry</name>
- <packaging>jar</packaging>
-
- <properties>
- <aws.sdk.version>1.12.276</aws.sdk.version>
- <aws.sdkv2.version>2.19.14</aws.sdkv2.version>
- <kotlin.version>1.7.10</kotlin.version>
- </properties>
-
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-annotations</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-metrics-core</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-base</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-base</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-common</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>bom</artifactId>
- <version>${aws.sdkv2.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
- <dependency>
- <groupId>com.amazonaws</groupId>
- <artifactId>aws-java-sdk-bom</artifactId>
- <version>${aws.sdk.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
-
- <!-- For dependency convergence -->
- <dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- <version>3.21.7</version>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>30.0-jre</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-reflect</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-common</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-jdk8</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-scripting-compiler-embeddable</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlinx</groupId>
- <artifactId>kotlinx-serialization-core-jvm</artifactId>
- <version>1.0.1</version>
- </dependency>
- <dependency>
- <groupId>org.checkerframework</groupId>
- <artifactId>checker-qual</artifactId>
- <version>3.5.0</version>
- </dependency>
- <dependency>
- <groupId>io.github.classgraph</groupId>
- <artifactId>classgraph</artifactId>
- <version>4.8.120</version>
- </dependency>
- <dependency>
- <groupId>com.squareup.okio</groupId>
- <artifactId>okio</artifactId>
- <version>2.8.0</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains</groupId>
- <artifactId>annotations</artifactId>
- <version>17.0.0</version>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
- <dependencies>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-end-to-end-tests-common</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-shaded-guava</artifactId>
- </dependency>
-
- <!-- Temporary: The tests need to be refactored to remove these dependencies -->
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kinesis</artifactId>
- <version>1.16.0</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-kinesis</artifactId>
- <version>1.16.0</version>
- <type>test-jar</type>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-kinesis-streams</artifactId>
- <version>1.16.0</version>
- <type>test-jar</type>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-base</artifactId>
- <version>${project.version}</version>
- <type>test-jar</type>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-clients</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-json-glue-schema-registry_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-surefire-plugin</artifactId>
- <configuration>
- <systemPropertyVariables>
- <!-- Required for Kinesalite. -->
- <!-- Including shaded and non-shaded conf to support test running from Maven and IntelliJ -->
- <com.amazonaws.sdk.disableCbor>true</com.amazonaws.sdk.disableCbor>
- <com.amazonaws.sdk.disableCertChecking>true</com.amazonaws.sdk.disableCertChecking>
- <org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCbor>true</org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCbor>
- <org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCertChecking>true</org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCertChecking>
- </systemPropertyVariables>
- </configuration>
- </plugin>
- </plugins>
- </build>
-
-</project>
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/java/org/apache/flink/glue/schema/registry/test/json/GSRKinesisPubsubClient.java b/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/java/org/apache/flink/glue/schema/registry/test/json/GSRKinesisPubsubClient.java
deleted file mode 100644
index 01668b803b2..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/java/org/apache/flink/glue/schema/registry/test/json/GSRKinesisPubsubClient.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * 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.glue.schema.registry.test.json;
-
-import org.apache.flink.streaming.connectors.kinesis.testutils.KinesisPubsubClient;
-
-import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
-import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-
-/**
- * Simple client to publish and retrieve messages, using the AWS Kinesis SDK, Flink Kinesis
- * Connectors and Glue Schema Registry classes.
- */
-public class GSRKinesisPubsubClient {
- private final KinesisPubsubClient client;
- private final GlueSchemaRegistrySerializationFacade serializationFacade;
- private final GlueSchemaRegistryDeserializationFacade deserializationFacade;
-
- public GSRKinesisPubsubClient(
- Properties properties, AwsCredentialsProvider gsrCredentialsProvider) {
- this.client = new KinesisPubsubClient(properties);
- this.serializationFacade = createSerializationFacade(gsrCredentialsProvider);
- this.deserializationFacade = createDeserializationFacade(gsrCredentialsProvider);
- }
-
- public void sendMessage(String schema, String streamName, Object msg) {
- UUID schemaVersionId =
- serializationFacade.getOrRegisterSchemaVersion(
- AWSSerializerInput.builder()
- .schemaDefinition(schema)
- .dataFormat(DataFormat.JSON.name())
- .schemaName(streamName)
- .transportName(streamName)
- .build());
-
- client.sendMessage(
- streamName, serializationFacade.serialize(DataFormat.JSON, msg, schemaVersionId));
- }
-
- public List<Object> readAllMessages(String streamName) throws Exception {
- return client.readAllMessages(
- streamName,
- bytes ->
- deserializationFacade.deserialize(
- AWSDeserializerInput.builder()
- .buffer(ByteBuffer.wrap(bytes))
- .transportName(streamName)
- .build()));
- }
-
- public void createStream(String stream, int shards, Properties props) throws Exception {
- client.createTopic(stream, shards, props);
- }
-
- private Map<String, Object> getSerDeConfigs() {
- Map<String, Object> configs = new HashMap<>();
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "ca-central-1");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-
- return configs;
- }
-
- private GlueSchemaRegistrySerializationFacade createSerializationFacade(
- final AwsCredentialsProvider credentialsProvider) {
- return GlueSchemaRegistrySerializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .glueSchemaRegistryConfiguration(
- new GlueSchemaRegistryConfiguration(getSerDeConfigs()))
- .build();
- }
-
- private GlueSchemaRegistryDeserializationFacade createDeserializationFacade(
- final AwsCredentialsProvider credentialsProvider) {
- return GlueSchemaRegistryDeserializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .configs(getSerDeConfigs())
- .build();
- }
-}
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/java/org/apache/flink/glue/schema/registry/test/json/GlueSchemaRegistryJsonKinesisITCase.java b/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/java/org/apache/flink/glue/schema/registry/test/json/GlueSchemaRegistryJsonKinesisITCase.java
deleted file mode 100644
index c9336b9eeb2..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/java/org/apache/flink/glue/schema/registry/test/json/GlueSchemaRegistryJsonKinesisITCase.java
+++ /dev/null
@@ -1,211 +0,0 @@
-/*
- * 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.glue.schema.registry.test.json;
-
-import org.apache.flink.api.common.time.Deadline;
-import org.apache.flink.connectors.kinesis.testutils.KinesaliteContainer;
-import org.apache.flink.formats.json.glue.schema.registry.GlueSchemaRegistryJsonDeserializationSchema;
-import org.apache.flink.formats.json.glue.schema.registry.GlueSchemaRegistryJsonSerializationSchema;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer;
-import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisProducer;
-import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants;
-import org.apache.flink.util.StringUtils;
-import org.apache.flink.util.TestLogger;
-
-import com.amazonaws.services.schemaregistry.serializers.json.JsonDataWithSchema;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import org.junit.After;
-import org.junit.Assume;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.Timeout;
-import org.testcontainers.containers.Network;
-import org.testcontainers.utility.DockerImageName;
-import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
-import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
-import software.amazon.awssdk.core.SdkSystemSetting;
-
-import java.net.URL;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-import static org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants.STREAM_INITIAL_POSITION;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** End-to-end test for Glue Schema Registry Json format using Kinesalite. */
-public class GlueSchemaRegistryJsonKinesisITCase extends TestLogger {
- private static final String INPUT_STREAM = "gsr_json_input_stream";
- private static final String OUTPUT_STREAM = "gsr_json_output_stream";
- private static final String INTER_CONTAINER_KINESALITE_ALIAS = "kinesalite";
- private static final String ACCESS_KEY = System.getenv("IT_CASE_GLUE_SCHEMA_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("IT_CASE_GLUE_SCHEMA_SECRET_KEY");
-
- private static final Network network = Network.newNetwork();
-
- @ClassRule public static final Timeout TIMEOUT = new Timeout(10, TimeUnit.MINUTES);
-
- @ClassRule
- public static final KinesaliteContainer KINESALITE =
- new KinesaliteContainer(
- DockerImageName.parse("instructure/kinesalite").withTag("latest"))
- .withNetwork(network)
- .withNetworkAliases(INTER_CONTAINER_KINESALITE_ALIAS);
-
- private GSRKinesisPubsubClient kinesisClient;
-
- @Before
- public void setUp() throws Exception {
- Assume.assumeTrue(
- "Access key not configured, skipping test...",
- !StringUtils.isNullOrWhitespaceOnly(ACCESS_KEY));
- Assume.assumeTrue(
- "Secret key not configured, skipping test...",
- !StringUtils.isNullOrWhitespaceOnly(SECRET_KEY));
-
- StaticCredentialsProvider gsrCredentialsProvider =
- StaticCredentialsProvider.create(
- AwsBasicCredentials.create(ACCESS_KEY, SECRET_KEY));
-
- Properties properties = KINESALITE.getContainerProperties();
-
- kinesisClient = new GSRKinesisPubsubClient(properties, gsrCredentialsProvider);
- kinesisClient.createStream(INPUT_STREAM, 2, properties);
- kinesisClient.createStream(OUTPUT_STREAM, 2, properties);
-
- System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");
- }
-
- @After
- public void teardown() {
- System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());
- }
-
- @Test
- public void testGSRJsonGenericFormatWithFlink() throws Exception {
-
- List<JsonDataWithSchema> messages = getGenericRecords();
- for (JsonDataWithSchema msg : messages) {
- kinesisClient.sendMessage(msg.getSchema(), INPUT_STREAM, msg);
- }
- log.info("generated records");
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
-
- DataStream<JsonDataWithSchema> input = env.addSource(createSource());
- input.addSink(createSink());
- env.executeAsync();
-
- Deadline deadline = Deadline.fromNow(Duration.ofSeconds(60));
- List<Object> results = kinesisClient.readAllMessages(OUTPUT_STREAM);
- while (deadline.hasTimeLeft() && results.size() < messages.size()) {
- log.info("waiting for results..");
- Thread.sleep(1000);
- results = kinesisClient.readAllMessages(OUTPUT_STREAM);
- }
- log.info("results: {}", results);
-
- assertThat(results).containsExactlyInAnyOrderElementsOf(messages);
- }
-
- private FlinkKinesisConsumer<JsonDataWithSchema> createSource() {
- Properties properties = KINESALITE.getContainerProperties();
- properties.setProperty(
- STREAM_INITIAL_POSITION,
- ConsumerConfigConstants.InitialPosition.TRIM_HORIZON.name());
-
- return new FlinkKinesisConsumer<>(
- INPUT_STREAM,
- new GlueSchemaRegistryJsonDeserializationSchema<>(
- JsonDataWithSchema.class, INPUT_STREAM, getConfigs()),
- properties);
- }
-
- private FlinkKinesisProducer<JsonDataWithSchema> createSink() throws Exception {
- FlinkKinesisProducer<JsonDataWithSchema> producer =
- new FlinkKinesisProducer<>(
- new GlueSchemaRegistryJsonSerializationSchema<>(
- OUTPUT_STREAM, getConfigs()),
- getProducerProperties());
- producer.setDefaultStream(OUTPUT_STREAM);
- producer.setDefaultPartition("fakePartition");
- return producer;
- }
-
- private Properties getProducerProperties() throws Exception {
- Properties producerProperties = KINESALITE.getContainerProperties();
- // producer needs region even when URL is specified
- producerProperties.put(ConsumerConfigConstants.AWS_REGION, "ca-central-1");
- // test driver does not deaggregate
- producerProperties.put("AggregationEnabled", String.valueOf(false));
-
- // KPL does not recognize endpoint URL..
- String kinesisUrl = producerProperties.getProperty(ConsumerConfigConstants.AWS_ENDPOINT);
- if (kinesisUrl != null) {
- URL url = new URL(kinesisUrl);
- producerProperties.put("KinesisEndpoint", url.getHost());
- producerProperties.put("KinesisPort", Integer.toString(url.getPort()));
- producerProperties.put("VerifyCertificate", "false");
- }
- return producerProperties;
- }
-
- private List<JsonDataWithSchema> getGenericRecords() {
- List<JsonDataWithSchema> records = new ArrayList<>();
- String schema =
- "{\"$id\":\"https://example.com/address.schema.json\","
- + "\"$schema\":\"http://json-schema.org/draft-07/schema#\","
- + "\"type\":\"object\","
- + "\"properties\":{"
- + "\"f1\":{\"type\":\"string\"},"
- + "\"f2\":{\"type\":\"integer\",\"maximum\":10000}}}";
- records.add(JsonDataWithSchema.builder(schema, "{\"f1\":\"olympic\",\"f2\":2020}").build());
- records.add(JsonDataWithSchema.builder(schema, "{\"f1\":\"iphone\",\"f2\":12}").build());
- records.add(
- JsonDataWithSchema.builder(schema, "{\"f1\":\"Stranger Things\",\"f2\":4}")
- .build());
- records.add(
- JsonDataWithSchema.builder(
- schema, "{\"f1\":\"Friends\",\"f2\":6,\"f3\":\"coming soon\"}")
- .build());
- records.add(JsonDataWithSchema.builder(schema, "{\"f1\":\"Porsche\",\"f2\":911}").build());
- records.add(JsonDataWithSchema.builder(schema, "{\"f1\":\"Ferrari\",\"f2\":488}").build());
- records.add(JsonDataWithSchema.builder(schema, "{\"f1\":\"McLaren\",\"f2\":720}").build());
- records.add(
- JsonDataWithSchema.builder(
- schema, "{\"f1\":\"Panorama\",\"f2\":360,\"f3\":\"Fantastic!\"}")
- .build());
- return records;
- }
-
- private Map<String, Object> getConfigs() {
- Map<String, Object> configs = new HashMap<>();
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "ca-central-1");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-
- return configs;
- }
-}
diff --git a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/resources/log4j2-test.properties b/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/resources/log4j2-test.properties
deleted file mode 100644
index 835c2ec9a3d..00000000000
--- a/flink-end-to-end-tests/flink-glue-schema-registry-json-test/src/test/resources/log4j2-test.properties
+++ /dev/null
@@ -1,28 +0,0 @@
-################################################################################
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-################################################################################
-
-# Set root logger level to OFF to not flood build logs
-# set manually to INFO for debugging purposes
-rootLogger.level = OFF
-rootLogger.appenderRef.test.ref = TestLogger
-
-appender.testlogger.name = TestLogger
-appender.testlogger.type = CONSOLE
-appender.testlogger.target = SYSTEM_ERR
-appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-end-to-end-tests/pom.xml b/flink-end-to-end-tests/pom.xml
index 3f9471487af..6c6d1be25a9 100644
--- a/flink-end-to-end-tests/pom.xml
+++ b/flink-end-to-end-tests/pom.xml
@@ -74,8 +74,6 @@ under the License.
<module>flink-netty-shuffle-memory-control-test</module>
<module>flink-python-test</module>
<module>flink-end-to-end-tests-hbase</module>
- <module>flink-glue-schema-registry-avro-test</module>
- <module>flink-glue-schema-registry-json-test</module>
<module>flink-end-to-end-tests-scala</module>
<module>flink-end-to-end-tests-sql</module>
</modules>
diff --git a/flink-formats/flink-avro-glue-schema-registry/archunit-violations/3e77c07d-dfdb-4ec0-8e64-5fad5c651c72 b/flink-formats/flink-avro-glue-schema-registry/archunit-violations/3e77c07d-dfdb-4ec0-8e64-5fad5c651c72
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/flink-formats/flink-avro-glue-schema-registry/archunit-violations/dc78e80c-3bb3-45bb-87c1-b57472d6f45b b/flink-formats/flink-avro-glue-schema-registry/archunit-violations/dc78e80c-3bb3-45bb-87c1-b57472d6f45b
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/flink-formats/flink-avro-glue-schema-registry/archunit-violations/stored.rules b/flink-formats/flink-avro-glue-schema-registry/archunit-violations/stored.rules
deleted file mode 100644
index 5354bcd9838..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/archunit-violations/stored.rules
+++ /dev/null
@@ -1,4 +0,0 @@
-#
-#Mon Apr 04 16:41:55 CEST 2022
-Tests\ inheriting\ from\ AbstractTestBase\ should\ have\ name\ ending\ with\ ITCase=3e77c07d-dfdb-4ec0-8e64-5fad5c651c72
-ITCASE\ tests\ should\ use\ a\ MiniCluster\ resource\ or\ extension=dc78e80c-3bb3-45bb-87c1-b57472d6f45b
diff --git a/flink-formats/flink-avro-glue-schema-registry/pom.xml b/flink-formats/flink-avro-glue-schema-registry/pom.xml
deleted file mode 100644
index c89e307cd25..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/pom.xml
+++ /dev/null
@@ -1,144 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <parent>
- <artifactId>flink-formats</artifactId>
- <groupId>org.apache.flink</groupId>
- <version>1.17-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
-
- <!-- There is a transitive scala dependency from mbknor-jackson-jsonschema -->
- <artifactId>flink-avro-glue-schema-registry_${scala.binary.version}</artifactId>
- <name>Flink : Formats : Avro AWS Glue Schema Registry</name>
- <packaging>jar</packaging>
-
- <properties>
- <glue.schema.registry.version>1.1.14</glue.schema.registry.version>
- <aws.sdkv2.version>2.19.14</aws.sdkv2.version>
- <kotlin.version>1.7.10</kotlin.version>
- </properties>
-
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>bom</artifactId>
- <version>${aws.sdkv2.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
-
- <!-- For dependency convergence -->
- <dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- <version>3.21.7</version>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>30.0-jre</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-reflect</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-common</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-jdk8</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-scripting-compiler-embeddable</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlinx</groupId>
- <artifactId>kotlinx-serialization-core-jvm</artifactId>
- <version>1.0.1</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains</groupId>
- <artifactId>annotations</artifactId>
- <version>17.0.0</version>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
- <dependencies>
- <!-- core dependencies -->
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-avro</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-base</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <!-- ArchUit test dependencies -->
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-architecture-tests-test</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <!-- This has a transitive dependency on mbknor-jackson-jsonschema which needs scala -->
- <groupId>software.amazon.glue</groupId>
- <artifactId>schema-registry-serde</artifactId>
- <version>${glue.schema.registry.version}</version>
- </dependency>
-
- <dependency>
- <groupId>software.amazon.glue</groupId>
- <artifactId>schema-registry-common</artifactId>
- <version>${glue.schema.registry.version}</version>
- </dependency>
-
- </dependencies>
-</project>
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroDeserializationSchema.java b/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroDeserializationSchema.java
deleted file mode 100644
index 75673a99fdf..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroDeserializationSchema.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.formats.avro.RegistryAvroDeserializationSchema;
-import org.apache.flink.formats.avro.SchemaCoder;
-
-import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.specific.SpecificRecord;
-
-import javax.annotation.Nullable;
-
-import java.util.Map;
-
-/**
- * AWS Glue Schema Registry Deserialization schema to de-serialize Avro binary format for Flink
- * Consumer user.
- *
- * @param <T> type of record it produces
- */
-@PublicEvolving
-public class GlueSchemaRegistryAvroDeserializationSchema<T>
- extends RegistryAvroDeserializationSchema<T> {
-
- /**
- * Creates a Avro deserialization schema.
- *
- * @param recordClazz class to which deserialize. Should be one of: {@link SpecificRecord},
- * {@link GenericRecord}.
- * @param reader reader's Avro schema. Should be provided if recordClazz is {@link
- * GenericRecord}
- * @param schemaCoderProvider schema coder provider which reads writer schema from AWS Glue
- * Schema Registry
- */
- private GlueSchemaRegistryAvroDeserializationSchema(
- Class<T> recordClazz,
- @Nullable Schema reader,
- SchemaCoder.SchemaCoderProvider schemaCoderProvider) {
- super(recordClazz, reader, schemaCoderProvider);
- }
-
- /**
- * Creates {@link GlueSchemaRegistryAvroDeserializationSchema} that produces {@link
- * GenericRecord} using provided schema.
- *
- * @param schema schema of produced records
- * @param configs configuration map of AWS Glue Schema Registry
- * @return deserialized record in form of {@link GenericRecord}
- */
- public static GlueSchemaRegistryAvroDeserializationSchema<GenericRecord> forGeneric(
- Schema schema, Map<String, Object> configs) {
- return new GlueSchemaRegistryAvroDeserializationSchema<>(
- GenericRecord.class,
- schema,
- new GlueSchemaRegistryAvroSchemaCoderProvider(configs));
- }
-
- /**
- * Creates {@link GlueSchemaRegistryAvroDeserializationSchema} that produces classes that were
- * generated from avro schema.
- *
- * @param clazz class of record to be produced
- * @param configs configuration map of AWS Glue Schema Registry
- * @return deserialized record
- */
- public static <T extends SpecificRecord>
- GlueSchemaRegistryAvroDeserializationSchema<T> forSpecific(
- Class<T> clazz, Map<String, Object> configs) {
- return new GlueSchemaRegistryAvroDeserializationSchema<>(
- clazz, null, new GlueSchemaRegistryAvroSchemaCoderProvider(configs));
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoder.java b/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoder.java
deleted file mode 100644
index 4dcfece1560..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoder.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.formats.avro.SchemaCoder;
-import org.apache.flink.util.Preconditions;
-
-import org.apache.avro.Schema;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.Map;
-
-/**
- * Schema coder that allows reading schema that is somehow embedded into serialized record. Used by
- * {@link GlueSchemaRegistryAvroDeserializationSchema} and {@link
- * GlueSchemaRegistryAvroSerializationSchema}.
- */
-@PublicEvolving
-public class GlueSchemaRegistryAvroSchemaCoder implements SchemaCoder {
- private GlueSchemaRegistryInputStreamDeserializer glueSchemaRegistryInputStreamDeserializer;
- private GlueSchemaRegistryOutputStreamSerializer glueSchemaRegistryOutputStreamSerializer;
-
- /**
- * Constructor accepts transport name and configuration map for AWS Glue Schema Registry.
- *
- * @param transportName topic name or stream name etc.
- * @param configs configurations for AWS Glue Schema Registry
- */
- public GlueSchemaRegistryAvroSchemaCoder(
- final String transportName, final Map<String, Object> configs) {
- glueSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(configs);
- glueSchemaRegistryOutputStreamSerializer =
- new GlueSchemaRegistryOutputStreamSerializer(transportName, configs);
- }
-
- @VisibleForTesting
- protected GlueSchemaRegistryAvroSchemaCoder(
- final GlueSchemaRegistryInputStreamDeserializer
- glueSchemaRegistryInputStreamDeserializer) {
- this.glueSchemaRegistryInputStreamDeserializer = glueSchemaRegistryInputStreamDeserializer;
- }
-
- @VisibleForTesting
- protected GlueSchemaRegistryAvroSchemaCoder(
- final GlueSchemaRegistryOutputStreamSerializer
- glueSchemaRegistryOutputStreamSerializer) {
- this.glueSchemaRegistryOutputStreamSerializer = glueSchemaRegistryOutputStreamSerializer;
- }
-
- @Override
- public Schema readSchema(InputStream in) throws IOException {
- return glueSchemaRegistryInputStreamDeserializer.getSchemaAndDeserializedStream(in);
- }
-
- @Override
- public void writeSchema(Schema schema, OutputStream out) throws IOException {
- Preconditions.checkArgument(
- out instanceof ByteArrayOutputStream, "The stream is not supported.");
- byte[] data = ((ByteArrayOutputStream) out).toByteArray();
- ((ByteArrayOutputStream) out).reset();
- glueSchemaRegistryOutputStreamSerializer.registerSchemaAndSerializeStream(
- schema, out, data);
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderProvider.java b/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderProvider.java
deleted file mode 100644
index d4c24f032a8..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderProvider.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.formats.avro.SchemaCoder;
-
-import java.util.Map;
-
-/** Provider for {@link GlueSchemaRegistryAvroSchemaCoder}. */
-@PublicEvolving
-public class GlueSchemaRegistryAvroSchemaCoderProvider implements SchemaCoder.SchemaCoderProvider {
- private final String transportName;
- private final Map<String, Object> configs;
-
- /**
- * Constructor used by {@link GlueSchemaRegistryAvroDeserializationSchema} and {@link
- * GlueSchemaRegistryAvroSerializationSchema}.
- *
- * @param transportName topic name or stream name etc.
- * @param configs configurations for AWS Glue Schema Registry
- */
- public GlueSchemaRegistryAvroSchemaCoderProvider(
- String transportName, Map<String, Object> configs) {
- this.transportName = transportName;
- this.configs = configs;
- }
-
- public GlueSchemaRegistryAvroSchemaCoderProvider(Map<String, Object> configs) {
- this(null, configs);
- }
-
- @Override
- public GlueSchemaRegistryAvroSchemaCoder get() {
- return new GlueSchemaRegistryAvroSchemaCoder(transportName, configs);
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSerializationSchema.java b/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSerializationSchema.java
deleted file mode 100644
index a68f3f92beb..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSerializationSchema.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.formats.avro.RegistryAvroSerializationSchema;
-import org.apache.flink.formats.avro.SchemaCoder;
-import org.apache.flink.util.WrappingRuntimeException;
-
-import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.io.Encoder;
-import org.apache.avro.specific.SpecificRecord;
-
-import javax.annotation.Nullable;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.Map;
-
-/**
- * AWS Glue Schema Registry Serialization schema to serialize to Avro binary format for Flink
- * Producer user.
- *
- * @param <T> the type to be serialized
- */
-@PublicEvolving
-public class GlueSchemaRegistryAvroSerializationSchema<T>
- extends RegistryAvroSerializationSchema<T> {
- /**
- * Creates an Avro serialization schema.
- *
- * @param recordClazz class to serialize. Should be one of: {@link SpecificRecord}, {@link
- * GenericRecord}.
- * @param reader reader's Avro schema. Should be provided if recordClazz is {@link
- * GenericRecord}
- * @param schemaCoderProvider schema coder provider which reads writer schema from AWS Glue
- * Schema Registry
- */
- private GlueSchemaRegistryAvroSerializationSchema(
- Class<T> recordClazz,
- @Nullable Schema reader,
- SchemaCoder.SchemaCoderProvider schemaCoderProvider) {
- super(recordClazz, reader, schemaCoderProvider);
- }
-
- @VisibleForTesting
- protected GlueSchemaRegistryAvroSerializationSchema(
- Class<T> recordClazz, @Nullable Schema reader, SchemaCoder schemaCoder) {
- // Pass null schema coder provider
- super(recordClazz, reader, null);
- this.schemaCoder = schemaCoder;
- }
-
- /**
- * Creates {@link GlueSchemaRegistryAvroSerializationSchema} that serializes {@link
- * GenericRecord} using provided schema.
- *
- * @param schema the schema that will be used for serialization
- * @param transportName topic name or stream name etc.
- * @param configs configuration map of AWS Glue Schema Registry
- * @return serialized record in form of byte array
- */
- public static GlueSchemaRegistryAvroSerializationSchema<GenericRecord> forGeneric(
- Schema schema, String transportName, Map<String, Object> configs) {
- return new GlueSchemaRegistryAvroSerializationSchema<>(
- GenericRecord.class,
- schema,
- new GlueSchemaRegistryAvroSchemaCoderProvider(transportName, configs));
- }
-
- /**
- * Creates {@link GlueSchemaRegistryAvroSerializationSchema} that serializes {@link
- * SpecificRecord} using provided schema.
- *
- * @param clazz the type to be serialized
- * @param transportName topic name or stream name etc.
- * @param configs configuration map of Amazon Schema Registry
- * @return serialized record in form of byte array
- */
- public static <T extends SpecificRecord>
- GlueSchemaRegistryAvroSerializationSchema<T> forSpecific(
- Class<T> clazz, String transportName, Map<String, Object> configs) {
- return new GlueSchemaRegistryAvroSerializationSchema<>(
- clazz, null, new GlueSchemaRegistryAvroSchemaCoderProvider(transportName, configs));
- }
-
- /**
- * Serializes the incoming element to a byte array containing bytes of AWS Glue Schema registry
- * information.
- *
- * @param object The incoming element to be serialized
- * @return The serialized bytes.
- */
- @Override
- public byte[] serialize(T object) {
- checkAvroInitialized();
-
- if (object == null) {
- return null;
- } else {
- try {
- ByteArrayOutputStream outputStream = getOutputStream();
- outputStream.reset();
- Encoder encoder = getEncoder();
- getDatumWriter().write(object, encoder);
- schemaCoder.writeSchema(getSchema(), outputStream);
- encoder.flush();
-
- return outputStream.toByteArray();
- } catch (IOException e) {
- throw new WrappingRuntimeException("Failed to serialize schema registry.", e);
- }
- }
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializer.java b/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializer.java
deleted file mode 100644
index 932910b5bf0..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializer.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.aws.util.AWSGeneralUtil;
-
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.exception.AWSSchemaRegistryException;
-import org.apache.avro.Schema;
-import org.apache.avro.Schema.Parser;
-import org.apache.avro.SchemaParseException;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Map;
-
-/**
- * AWS Glue Schema Registry input stream de-serializer to accept input stream and extract schema
- * from it and remove schema registry information in the input stream.
- */
-@PublicEvolving
-public class GlueSchemaRegistryInputStreamDeserializer {
- private final GlueSchemaRegistryDeserializationFacade glueSchemaRegistryDeserializationFacade;
-
- /**
- * Constructor accepts configuration map for AWS Deserializer.
- *
- * @param configs configuration map
- */
- public GlueSchemaRegistryInputStreamDeserializer(Map<String, Object> configs) {
- AwsCredentialsProvider credentialsProvider = AWSGeneralUtil.getCredentialsProvider(configs);
-
- this.glueSchemaRegistryDeserializationFacade =
- GlueSchemaRegistryDeserializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .configs(configs)
- .build();
- }
-
- public GlueSchemaRegistryInputStreamDeserializer(
- GlueSchemaRegistryDeserializationFacade glueSchemaRegistryDeserializationFacade) {
- this.glueSchemaRegistryDeserializationFacade = glueSchemaRegistryDeserializationFacade;
- }
-
- /**
- * Get schema and remove extra Schema Registry information within input stream.
- *
- * @param in input stream
- * @return schema of object within input stream
- * @throws IOException Exception during decompression
- */
- public Schema getSchemaAndDeserializedStream(InputStream in) throws IOException {
- byte[] inputBytes = new byte[in.available()];
- in.read(inputBytes);
- in.reset();
-
- String schemaDefinition =
- glueSchemaRegistryDeserializationFacade.getSchemaDefinition(inputBytes);
-
- Schema schema;
- try {
- Parser schemaParser = new Schema.Parser();
- schema = schemaParser.parse(schemaDefinition);
- } catch (SchemaParseException e) {
- String message =
- "Error occurred while parsing schema, see inner exception for details.";
- throw new AWSSchemaRegistryException(message, e);
- }
-
- return schema;
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryOutputStreamSerializer.java b/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryOutputStreamSerializer.java
deleted file mode 100644
index 96e3efe634f..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/main/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryOutputStreamSerializer.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.connector.aws.util.AWSGeneralUtil;
-
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.GlueSchemaRegistryUtils;
-import org.apache.avro.Schema;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.Map;
-
-/**
- * AWS Glue Schema Registry output stream serializer to accept schema and output stream to register
- * schema and write serialized object with schema registry bytes to output stream.
- */
-@PublicEvolving
-public class GlueSchemaRegistryOutputStreamSerializer {
- private final String transportName;
- private final Map<String, Object> configs;
- private final GlueSchemaRegistrySerializationFacade glueSchemaRegistrySerializationFacade;
-
- public GlueSchemaRegistryOutputStreamSerializer(
- String transportName, Map<String, Object> configs) {
- this(transportName, configs, null);
- }
-
- public GlueSchemaRegistryOutputStreamSerializer(
- String transportName,
- Map<String, Object> configs,
- GlueSchemaRegistrySerializationFacade glueSchemaRegistrySerializationFacade) {
- this.transportName = transportName;
- this.configs = configs;
-
- AwsCredentialsProvider credentialsProvider = AWSGeneralUtil.getCredentialsProvider(configs);
-
- this.glueSchemaRegistrySerializationFacade =
- glueSchemaRegistrySerializationFacade != null
- ? glueSchemaRegistrySerializationFacade
- : GlueSchemaRegistrySerializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .glueSchemaRegistryConfiguration(
- new GlueSchemaRegistryConfiguration(configs))
- .build();
- }
-
- /**
- * Register schema and write serialized object with schema registry bytes to output stream.
- *
- * @param schema schema to be registered
- * @param out output stream
- * @param data original bytes of serialized object
- * @throws IOException
- */
- public void registerSchemaAndSerializeStream(Schema schema, OutputStream out, byte[] data)
- throws IOException {
- byte[] bytes =
- glueSchemaRegistrySerializationFacade.encode(
- transportName,
- new com.amazonaws.services.schemaregistry.common.Schema(
- schema.toString(), DataFormat.AVRO.name(), getSchemaName()),
- data);
- out.write(bytes);
- }
-
- private String getSchemaName() {
- String schemaName = GlueSchemaRegistryUtils.getInstance().getSchemaName(configs);
-
- return schemaName != null
- ? schemaName
- : GlueSchemaRegistryUtils.getInstance()
- .configureSchemaNamingStrategy(configs)
- .getSchemaName(transportName);
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/architecture/TestCodeArchitectureTest.java b/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/architecture/TestCodeArchitectureTest.java
deleted file mode 100644
index f0cc3d795e3..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/architecture/TestCodeArchitectureTest.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.architecture;
-
-import org.apache.flink.architecture.common.ImportOptions;
-
-import com.tngtech.archunit.core.importer.ImportOption;
-import com.tngtech.archunit.junit.AnalyzeClasses;
-import com.tngtech.archunit.junit.ArchTest;
-import com.tngtech.archunit.junit.ArchTests;
-
-/** Architecture tests for test code. */
-@AnalyzeClasses(
- packages = {"org.apache.flink.formats.avro.glue.schema.registry"},
- importOptions = {
- ImportOption.OnlyIncludeTests.class,
- ImportOptions.ExcludeScalaImportOption.class,
- ImportOptions.ExcludeShadedImportOption.class
- })
-public class TestCodeArchitectureTest {
-
- @ArchTest
- public static final ArchTests COMMON_TESTS = ArchTests.in(TestCodeArchitectureTestBase.class);
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroDeserializationSchemaTest.java b/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroDeserializationSchemaTest.java
deleted file mode 100644
index 6a95b5453d6..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroDeserializationSchemaTest.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import org.apache.avro.Schema;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link GlueSchemaRegistryAvroDeserializationSchema}. */
-class GlueSchemaRegistryAvroDeserializationSchemaTest {
- private static final String AVRO_USER_SCHEMA_FILE = "src/test/java/resources/avro/user.avsc";
- private static Schema userSchema;
- private static Map<String, Object> configs = new HashMap<>();
-
- @BeforeAll
- static void setup() throws IOException {
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-
- Schema.Parser parser = new Schema.Parser();
- userSchema = parser.parse(new File(AVRO_USER_SCHEMA_FILE));
- }
-
- /** Test whether forGeneric method works. */
- @Test
- void testForGeneric_withValidParams_succeeds() {
- assertThat(GlueSchemaRegistryAvroDeserializationSchema.forGeneric(userSchema, configs))
- .isNotNull()
- .isInstanceOf(GlueSchemaRegistryAvroDeserializationSchema.class);
- }
-
- /** Test whether forSpecific method works. */
- @Test
- void testForSpecific_withValidParams_succeeds() {
- assertThat(GlueSchemaRegistryAvroDeserializationSchema.forSpecific(User.class, configs))
- .isNotNull()
- .isInstanceOf(GlueSchemaRegistryAvroDeserializationSchema.class);
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderTest.java b/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderTest.java
deleted file mode 100644
index b05eaa7a71b..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderTest.java
+++ /dev/null
@@ -1,221 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import com.amazonaws.services.schemaregistry.common.AWSSchemaRegistryClient;
-import com.amazonaws.services.schemaregistry.common.SchemaByDefinitionFetcher;
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.exception.AWSSchemaRegistryException;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import lombok.NonNull;
-import org.apache.avro.Schema;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.EntityNotFoundException;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** Tests for {@link GlueSchemaRegistryAvroSchemaCoder}. */
-class GlueSchemaRegistryAvroSchemaCoderTest {
- private static final String testTopic = "Test-Topic";
- private static final String schemaName = "User-Topic";
- private static final String AVRO_USER_SCHEMA_FILE = "src/test/java/resources/avro/user.avsc";
- private static final byte[] actualBytes =
- new byte[] {12, 99, 8, 116, 101, 115, 116, 0, 20, 0, 12, 118, 105, 111, 108, 101, 116};
- private static final byte[] specificBytes =
- new byte[] {
- 3, 0, -73, -76, -89, -16, -100, -106, 78, 74, -90, -121, -5, 93, -23, -17, 12, 99,
- 8, 116, 101, 115, 116, 0, 20, 0, 12, 118, 105, 111, 108, 101, 116
- };
- private static final Map<String, Object> configs = new HashMap<>();
- private static final Map<String, String> metadata = new HashMap<>();
- private static final AwsCredentialsProvider credentialsProvider =
- DefaultCredentialsProvider.builder().build();
- private static Schema userSchema;
- private static User userDefinedPojo;
- private static GlueSchemaRegistryInputStreamDeserializer mockInputStreamDeserializer;
- private static GlueSchemaRegistryOutputStreamSerializer mockOutputStreamSerializer;
-
- @BeforeAll
- static void setup() throws IOException {
- metadata.put("test-key", "test-value");
- metadata.put(AWSSchemaRegistryConstants.TRANSPORT_METADATA_KEY, testTopic);
-
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
- configs.put(AWSSchemaRegistryConstants.SCHEMA_NAME, schemaName);
- configs.put(AWSSchemaRegistryConstants.METADATA, metadata);
-
- Schema.Parser parser = new Schema.Parser();
- userSchema = parser.parse(new File(AVRO_USER_SCHEMA_FILE));
- userDefinedPojo =
- User.newBuilder()
- .setName("test_avro_schema")
- .setFavoriteColor("violet")
- .setFavoriteNumber(10)
- .build();
-
- mockInputStreamDeserializer = new MockGlueSchemaRegistryInputStreamDeserializer();
- mockOutputStreamSerializer = new MockGlueSchemaRegistryOutputStreamSerializer();
- }
-
- /** Test whether constructor works. */
- @Test
- void testConstructor_withConfigs_succeeds() {
- assertThat(new GlueSchemaRegistryAvroSchemaCoder(testTopic, configs)).isNotNull();
- }
-
- /** Test whether readSchema method works. */
- @Test
- void testReadSchema_withValidParams_succeeds() throws IOException {
- GlueSchemaRegistryAvroSchemaCoder glueSchemaRegistryAvroSchemaCoder =
- new GlueSchemaRegistryAvroSchemaCoder(mockInputStreamDeserializer);
- Schema resultSchema =
- glueSchemaRegistryAvroSchemaCoder.readSchema(buildByteArrayInputStream());
-
- assertThat(resultSchema).isEqualTo(userSchema);
- }
-
- /** Test whether writeSchema method works. */
- @Test
- void testWriteSchema_withValidParams_succeeds() throws IOException {
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
- outputStream.write(actualBytes);
- GlueSchemaRegistryAvroSchemaCoder glueSchemaRegistryAvroSchemaCoder =
- new GlueSchemaRegistryAvroSchemaCoder(mockOutputStreamSerializer);
- glueSchemaRegistryAvroSchemaCoder.writeSchema(userSchema, outputStream);
-
- testForSerializedData(outputStream.toByteArray());
- }
-
- /** Test whether writeSchema method throws exception if auto registration un-enabled. */
- @Test
- void testWriteSchema_withoutAutoRegistration_throwsException() {
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, false);
- SchemaByDefinitionFetcher fetcher =
- new SchemaByDefinitionFetcher(
- new MockAWSSchemaRegistryClient(),
- new GlueSchemaRegistryConfiguration(configs));
-
- GlueSchemaRegistrySerializationFacade glueSchemaRegistrySerializationFacade =
- GlueSchemaRegistrySerializationFacade.builder()
- .schemaByDefinitionFetcher(fetcher)
- .credentialProvider(credentialsProvider)
- .glueSchemaRegistryConfiguration(
- new GlueSchemaRegistryConfiguration(configs))
- .build();
-
- GlueSchemaRegistryOutputStreamSerializer glueSchemaRegistryOutputStreamSerializer =
- new GlueSchemaRegistryOutputStreamSerializer(
- testTopic, configs, glueSchemaRegistrySerializationFacade);
- GlueSchemaRegistryAvroSchemaCoder glueSchemaRegistryAvroSchemaCoder =
- new GlueSchemaRegistryAvroSchemaCoder(glueSchemaRegistryOutputStreamSerializer);
-
- assertThatThrownBy(
- () ->
- glueSchemaRegistryAvroSchemaCoder.writeSchema(
- userSchema, new ByteArrayOutputStream()))
- .isInstanceOf(AWSSchemaRegistryException.class)
- .hasMessage(AWSSchemaRegistryConstants.AUTO_REGISTRATION_IS_DISABLED_MSG);
- }
-
- private void testForSerializedData(byte[] serializedData) {
- assertThat(serializedData).isNotNull();
-
- ByteBuffer buffer = getByteBuffer(serializedData);
- byte headerVersionByte = getByte(buffer);
-
- assertThat(headerVersionByte).isEqualTo(AWSSchemaRegistryConstants.HEADER_VERSION_BYTE);
- }
-
- private ByteArrayInputStream buildByteArrayInputStream() {
- return new ByteArrayInputStream(specificBytes);
- }
-
- private ByteBuffer getByteBuffer(byte[] bytes) {
- return ByteBuffer.wrap(bytes);
- }
-
- private Byte getByte(ByteBuffer buffer) {
- return buffer.get();
- }
-
- private static class MockGlueSchemaRegistryInputStreamDeserializer
- extends GlueSchemaRegistryInputStreamDeserializer {
-
- public MockGlueSchemaRegistryInputStreamDeserializer() {
- super((GlueSchemaRegistryDeserializationFacade) null);
- }
-
- @Override
- public Schema getSchemaAndDeserializedStream(InputStream in) {
- return userSchema;
- }
- }
-
- private static class MockGlueSchemaRegistryOutputStreamSerializer
- extends GlueSchemaRegistryOutputStreamSerializer {
-
- public MockGlueSchemaRegistryOutputStreamSerializer() {
- super(testTopic, configs, null);
- }
-
- @Override
- public void registerSchemaAndSerializeStream(Schema schema, OutputStream out, byte[] data)
- throws IOException {
- out.write(specificBytes);
- }
- }
-
- private static class MockAWSSchemaRegistryClient extends AWSSchemaRegistryClient {
-
- public MockAWSSchemaRegistryClient() {
- super(credentialsProvider, new GlueSchemaRegistryConfiguration(configs));
- }
-
- @Override
- public UUID getSchemaVersionIdByDefinition(
- @NonNull String schemaDefinition,
- @NonNull String schemaName,
- @NonNull String dataFormat) {
- EntityNotFoundException entityNotFoundException =
- EntityNotFoundException.builder()
- .message(AWSSchemaRegistryConstants.SCHEMA_NOT_FOUND_MSG)
- .build();
- throw new AWSSchemaRegistryException(entityNotFoundException);
- }
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSerializationSchemaTest.java b/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSerializationSchemaTest.java
deleted file mode 100644
index aede17bc063..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSerializationSchemaTest.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import org.apache.avro.Schema;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link GlueSchemaRegistryAvroSerializationSchema}. */
-class GlueSchemaRegistryAvroSerializationSchemaTest {
- private static final String testTopic = "Test-Topic";
- private static final String schemaName = "User-Topic";
- private static final String AVRO_USER_SCHEMA_FILE = "src/test/java/resources/avro/user.avsc";
- private static final byte[] specificBytes =
- new byte[] {
- 3, 0, -73, -76, -89, -16, -100, -106, 78, 74, -90, -121, -5, 93, -23, -17, 12, 99,
- 8, 116, 101, 115, 116, 0, 20, 0, 12, 118, 105, 111, 108, 101, 116
- };
- private static Schema userSchema;
- private static User userDefinedPojo;
- private static final Map<String, Object> configs = new HashMap<>();
- private static final Map<String, String> metadata = new HashMap<>();
- private static GlueSchemaRegistryConfiguration glueSchemaRegistryConfiguration;
- private static final AwsCredentialsProvider credentialsProvider =
- DefaultCredentialsProvider.builder().build();
- private static GlueSchemaRegistrySerializationFacade mockSerializationFacade;
-
- @BeforeAll
- static void setup() throws IOException {
- metadata.put("test-key", "test-value");
- metadata.put(AWSSchemaRegistryConstants.TRANSPORT_METADATA_KEY, testTopic);
-
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
- configs.put(AWSSchemaRegistryConstants.SCHEMA_NAME, schemaName);
- configs.put(AWSSchemaRegistryConstants.METADATA, metadata);
-
- Schema.Parser parser = new Schema.Parser();
- userSchema = parser.parse(new File(AVRO_USER_SCHEMA_FILE));
- userDefinedPojo =
- User.newBuilder()
- .setName("test_avro_schema")
- .setFavoriteColor("violet")
- .setFavoriteNumber(10)
- .build();
-
- glueSchemaRegistryConfiguration = new GlueSchemaRegistryConfiguration(configs);
- mockSerializationFacade = new MockGlueSchemaRegistrySerializationFacade();
- }
-
- /** Test whether forGeneric method works. */
- @Test
- void testForGeneric_withValidParams_succeeds() {
- assertThat(
- GlueSchemaRegistryAvroSerializationSchema.forGeneric(
- userSchema, testTopic, configs))
- .isNotNull();
- assertThat(
- GlueSchemaRegistryAvroSerializationSchema.forGeneric(
- userSchema, testTopic, configs))
- .isInstanceOf(GlueSchemaRegistryAvroSerializationSchema.class);
- }
-
- /** Test whether forSpecific method works. */
- @Test
- void testForSpecific_withValidParams_succeeds() {
- assertThat(
- GlueSchemaRegistryAvroSerializationSchema.forSpecific(
- User.class, testTopic, configs))
- .isNotNull();
- assertThat(
- GlueSchemaRegistryAvroSerializationSchema.forSpecific(
- User.class, testTopic, configs))
- .isInstanceOf(GlueSchemaRegistryAvroSerializationSchema.class);
- }
-
- /** Test whether serialize method when compression is not enabled works. */
- @Test
- void testSerialize_withValidParams_withoutCompression_succeeds() {
- AWSSchemaRegistryConstants.COMPRESSION compressionType =
- AWSSchemaRegistryConstants.COMPRESSION.NONE;
- configs.put(AWSSchemaRegistryConstants.COMPRESSION_TYPE, compressionType.name());
-
- GlueSchemaRegistryOutputStreamSerializer glueSchemaRegistryOutputStreamSerializer =
- new GlueSchemaRegistryOutputStreamSerializer(
- testTopic, configs, mockSerializationFacade);
- GlueSchemaRegistryAvroSchemaCoder glueSchemaRegistryAvroSchemaCoder =
- new GlueSchemaRegistryAvroSchemaCoder(glueSchemaRegistryOutputStreamSerializer);
- GlueSchemaRegistryAvroSerializationSchema<User> glueSchemaRegistryAvroSerializationSchema =
- new GlueSchemaRegistryAvroSerializationSchema<>(
- User.class, null, glueSchemaRegistryAvroSchemaCoder);
-
- byte[] serializedData =
- glueSchemaRegistryAvroSerializationSchema.serialize(userDefinedPojo);
- assertThat(serializedData).isEqualTo(specificBytes);
- }
-
- /** Test whether serialize method when compression is enabled works. */
- @Test
- void testSerialize_withValidParams_withCompression_succeeds() {
- AWSSchemaRegistryConstants.COMPRESSION compressionType =
- AWSSchemaRegistryConstants.COMPRESSION.ZLIB;
- configs.put(AWSSchemaRegistryConstants.COMPRESSION_TYPE, compressionType.name());
-
- GlueSchemaRegistryOutputStreamSerializer glueSchemaRegistryOutputStreamSerializer =
- new GlueSchemaRegistryOutputStreamSerializer(
- testTopic, configs, mockSerializationFacade);
- GlueSchemaRegistryAvroSchemaCoder glueSchemaRegistryAvroSchemaCoder =
- new GlueSchemaRegistryAvroSchemaCoder(glueSchemaRegistryOutputStreamSerializer);
- GlueSchemaRegistryAvroSerializationSchema<User> glueSchemaRegistryAvroSerializationSchema =
- new GlueSchemaRegistryAvroSerializationSchema<>(
- User.class, null, glueSchemaRegistryAvroSchemaCoder);
-
- byte[] serializedData =
- glueSchemaRegistryAvroSerializationSchema.serialize(userDefinedPojo);
- assertThat(serializedData).isEqualTo(specificBytes);
- }
-
- /** Test whether serialize method returns null when input object is null. */
- @Test
- void testSerialize_withNullObject_returnNull() {
- GlueSchemaRegistryAvroSerializationSchema<User> glueSchemaRegistryAvroSerializationSchema =
- GlueSchemaRegistryAvroSerializationSchema.forSpecific(
- User.class, testTopic, configs);
- assertThat(glueSchemaRegistryAvroSerializationSchema.serialize(null)).isNull();
- }
-
- private static class MockGlueSchemaRegistrySerializationFacade
- extends GlueSchemaRegistrySerializationFacade {
-
- public MockGlueSchemaRegistrySerializationFacade() {
- super(credentialsProvider, null, glueSchemaRegistryConfiguration, configs, null);
- }
-
- @Override
- public byte[] encode(
- String transportName,
- com.amazonaws.services.schemaregistry.common.Schema schema,
- byte[] data) {
- return specificBytes;
- }
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializerTest.java b/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializerTest.java
deleted file mode 100644
index 682e7bfc1f7..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializerTest.java
+++ /dev/null
@@ -1,311 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.flink.formats.avro.utils.MutableByteArrayInputStream;
-
-import com.amazonaws.services.schemaregistry.common.GlueSchemaRegistryCompressionHandler;
-import com.amazonaws.services.schemaregistry.common.GlueSchemaRegistryDefaultCompression;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.exception.AWSSchemaRegistryException;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants.COMPRESSION;
-import lombok.NonNull;
-import org.apache.avro.Schema;
-import org.apache.avro.io.BinaryEncoder;
-import org.apache.avro.io.DatumWriter;
-import org.apache.avro.io.EncoderFactory;
-import org.apache.avro.specific.SpecificDatumWriter;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
-import static com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants.COMPRESSION.NONE;
-import static com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants.COMPRESSION_DEFAULT_BYTE;
-import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_ACCESS_KEY_ID;
-import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_SECRET_ACCESS_KEY;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** Tests for {@link GlueSchemaRegistryInputStreamDeserializer}. */
-class GlueSchemaRegistryInputStreamDeserializerTest {
- private static final String testTopic = "Test-Topic";
- private static final UUID USER_SCHEMA_VERSION_ID = UUID.randomUUID();
- private static final String AVRO_USER_SCHEMA_FILE = "src/test/java/resources/avro/user.avsc";
- private static byte compressionByte;
- private static Schema userSchema;
- private static com.amazonaws.services.schemaregistry.common.Schema glueSchema;
- private static User userDefinedPojo;
- private static final Map<String, Object> configs = new HashMap<>();
- private static final Map<String, String> metadata = new HashMap<>();
- private static GlueSchemaRegistryCompressionHandler compressionHandler;
- private static final AwsCredentialsProvider credentialsProvider =
- DefaultCredentialsProvider.builder().build();
- private GlueSchemaRegistryDeserializationFacade glueSchemaRegistryDeserializationFacade;
-
- @BeforeEach
- void setup() throws IOException {
- metadata.put("test-key", "test-value");
- metadata.put(AWSSchemaRegistryConstants.TRANSPORT_METADATA_KEY, testTopic);
-
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-
- Schema.Parser parser = new Schema.Parser();
- userSchema = parser.parse(new File(AVRO_USER_SCHEMA_FILE));
- glueSchema =
- new com.amazonaws.services.schemaregistry.common.Schema(
- userSchema.toString(), DataFormat.AVRO.name(), testTopic);
- userDefinedPojo =
- User.newBuilder()
- .setName("test_avro_schema")
- .setFavoriteColor("violet")
- .setFavoriteNumber(10)
- .build();
- }
-
- /** Test whether constructor works with configuration map. */
- @Test
- void testConstructor_withConfigs_succeeds() {
- GlueSchemaRegistryInputStreamDeserializer glueSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(configs);
- assertThat(glueSchemaRegistryInputStreamDeserializer)
- .isInstanceOf(GlueSchemaRegistryInputStreamDeserializer.class);
- }
-
- @Test
- void testDefaultAwsCredentialsProvider() throws Exception {
- GlueSchemaRegistryInputStreamDeserializer glueSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(configs);
-
- GlueSchemaRegistryDeserializationFacade facade =
- getField(
- "glueSchemaRegistryDeserializationFacade",
- glueSchemaRegistryInputStreamDeserializer);
-
- AwsCredentialsProvider credentialsProvider = facade.getCredentialsProvider();
- assertThat(credentialsProvider).isInstanceOf(DefaultCredentialsProvider.class);
- }
-
- @Test
- void testAwsCredentialsProviderFromConfig() throws Exception {
- Map<String, Object> config = new HashMap<>(configs);
- config.put(AWS_ACCESS_KEY_ID, "ak");
- config.put(AWS_SECRET_ACCESS_KEY, "sk");
-
- GlueSchemaRegistryInputStreamDeserializer glueSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(config);
-
- GlueSchemaRegistryDeserializationFacade facade =
- getField(
- "glueSchemaRegistryDeserializationFacade",
- glueSchemaRegistryInputStreamDeserializer);
-
- AwsCredentialsProvider credentialsProvider = facade.getCredentialsProvider();
- assertThat(credentialsProvider.resolveCredentials().accessKeyId()).isEqualTo("ak");
- assertThat(credentialsProvider.resolveCredentials().secretAccessKey()).isEqualTo("sk");
- }
-
- /** Test whether constructor works with AWS de-serializer input. */
- @Test
- void testConstructor_withDeserializer_succeeds() {
- GlueSchemaRegistryInputStreamDeserializer glueSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(
- glueSchemaRegistryDeserializationFacade);
- assertThat(glueSchemaRegistryInputStreamDeserializer)
- .isInstanceOf(GlueSchemaRegistryInputStreamDeserializer.class);
- }
-
- /** Test whether getSchemaAndDeserializedStream method when compression is not enabled works. */
- @Test
- void testGetSchemaAndDeserializedStream_withoutCompression_succeeds() throws IOException {
- compressionByte = COMPRESSION_DEFAULT_BYTE;
- compressionHandler = new GlueSchemaRegistryDefaultCompression();
-
- ByteArrayOutputStream byteArrayOutputStream =
- buildByteArrayOutputStream(
- AWSSchemaRegistryConstants.HEADER_VERSION_BYTE, compressionByte);
- byte[] bytes =
- writeToExistingStream(
- byteArrayOutputStream,
- encodeData(userDefinedPojo, new SpecificDatumWriter<>(userSchema)));
-
- MutableByteArrayInputStream mutableByteArrayInputStream = new MutableByteArrayInputStream();
- mutableByteArrayInputStream.setBuffer(bytes);
- glueSchemaRegistryDeserializationFacade =
- new MockGlueSchemaRegistryDeserializationFacade(bytes, glueSchema, NONE);
-
- GlueSchemaRegistryInputStreamDeserializer glueSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(
- glueSchemaRegistryDeserializationFacade);
- Schema resultSchema =
- glueSchemaRegistryInputStreamDeserializer.getSchemaAndDeserializedStream(
- mutableByteArrayInputStream);
-
- assertThat(resultSchema.toString()).isEqualTo(glueSchema.getSchemaDefinition());
- }
-
- /** Test whether getSchemaAndDeserializedStream method when compression is enabled works. */
- @Test
- void testGetSchemaAndDeserializedStream_withCompression_succeeds() throws IOException {
- COMPRESSION compressionType = COMPRESSION.ZLIB;
- compressionByte = AWSSchemaRegistryConstants.COMPRESSION_BYTE;
- compressionHandler = new GlueSchemaRegistryDefaultCompression();
-
- ByteArrayOutputStream byteArrayOutputStream =
- buildByteArrayOutputStream(
- AWSSchemaRegistryConstants.HEADER_VERSION_BYTE, compressionByte);
- byte[] bytes =
- writeToExistingStream(
- byteArrayOutputStream,
- compressData(
- encodeData(
- userDefinedPojo, new SpecificDatumWriter<>(userSchema))));
-
- MutableByteArrayInputStream mutableByteArrayInputStream = new MutableByteArrayInputStream();
- mutableByteArrayInputStream.setBuffer(bytes);
- glueSchemaRegistryDeserializationFacade =
- new MockGlueSchemaRegistryDeserializationFacade(bytes, glueSchema, compressionType);
-
- GlueSchemaRegistryInputStreamDeserializer glueSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(
- glueSchemaRegistryDeserializationFacade);
- Schema resultSchema =
- glueSchemaRegistryInputStreamDeserializer.getSchemaAndDeserializedStream(
- mutableByteArrayInputStream);
-
- assertThat(resultSchema.toString()).isEqualTo(glueSchema.getSchemaDefinition());
- }
-
- /** Test whether getSchemaAndDeserializedStream method throws exception with invalid schema. */
- @Test
- void testGetSchemaAndDeserializedStream_withWrongSchema_throwsException() {
- String schemaDefinition =
- "{"
- + "\"type\":\"record\","
- + "\"name\":\"User\","
- + "\"namespace\":\"org.apache.flink.formats.avro.glue.schema.registry\","
- + "\"fields\":"
- + "["
- + "{\"name\":\"name\",\"type\":\"string\"},"
- + "{\"name\":\"favorite_number\",\"name\":[\"int\",\"null\"]},"
- + "{\"name\":\"favorite_color\",\"type\":[\"string\",\"null\"]}"
- + "]"
- + "}";
- MutableByteArrayInputStream mutableByteArrayInputStream = new MutableByteArrayInputStream();
- glueSchema =
- new com.amazonaws.services.schemaregistry.common.Schema(
- schemaDefinition, DataFormat.AVRO.name(), testTopic);
- glueSchemaRegistryDeserializationFacade =
- new MockGlueSchemaRegistryDeserializationFacade(new byte[20], glueSchema, NONE);
- GlueSchemaRegistryInputStreamDeserializer awsSchemaRegistryInputStreamDeserializer =
- new GlueSchemaRegistryInputStreamDeserializer(
- glueSchemaRegistryDeserializationFacade);
-
- assertThatThrownBy(
- () ->
- awsSchemaRegistryInputStreamDeserializer
- .getSchemaAndDeserializedStream(
- mutableByteArrayInputStream))
- .isInstanceOf(AWSSchemaRegistryException.class)
- .hasMessage(
- "Error occurred while parsing schema, see inner exception for details.");
- }
-
- private ByteArrayOutputStream buildByteArrayOutputStream(byte headerByte, byte compressionByte)
- throws IOException {
- ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
- byteArrayOutputStream.write(headerByte);
- byteArrayOutputStream.write(compressionByte);
- writeSchemaVersionId(byteArrayOutputStream);
-
- return byteArrayOutputStream;
- }
-
- private void writeSchemaVersionId(ByteArrayOutputStream out) throws IOException {
- ByteBuffer buffer =
- ByteBuffer.wrap(new byte[AWSSchemaRegistryConstants.SCHEMA_VERSION_ID_SIZE]);
- buffer.putLong(USER_SCHEMA_VERSION_ID.getMostSignificantBits());
- buffer.putLong(USER_SCHEMA_VERSION_ID.getLeastSignificantBits());
- out.write(buffer.array());
- }
-
- private byte[] writeToExistingStream(ByteArrayOutputStream toStream, byte[] fromStream)
- throws IOException {
- toStream.write(fromStream);
- return toStream.toByteArray();
- }
-
- private byte[] encodeData(Object object, DatumWriter<Object> writer) throws IOException {
- ByteArrayOutputStream actualDataBytes = new ByteArrayOutputStream();
- BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(actualDataBytes, null);
- writer.write(object, encoder);
- encoder.flush();
- return actualDataBytes.toByteArray();
- }
-
- private byte[] compressData(byte[] actualDataBytes) throws IOException {
- return compressionHandler.compress(actualDataBytes);
- }
-
- private static class MockGlueSchemaRegistryDeserializationFacade
- extends GlueSchemaRegistryDeserializationFacade {
- private final byte[] bytes;
- private final com.amazonaws.services.schemaregistry.common.Schema schema;
- private final COMPRESSION compressionType;
-
- public MockGlueSchemaRegistryDeserializationFacade(
- byte[] bytes,
- com.amazonaws.services.schemaregistry.common.Schema schema,
- COMPRESSION compressionType) {
- super(configs, null, credentialsProvider, null);
- this.bytes = bytes;
- this.schema = schema;
- this.compressionType = compressionType;
- }
-
- @Override
- public String getSchemaDefinition(@NonNull byte[] data) {
- return schema.getSchemaDefinition();
- }
-
- @Override
- public byte[] getActualData(byte[] data) {
- return bytes;
- }
- }
-
- private <T> T getField(final String fieldName, final Object instance) throws Exception {
- Field field = instance.getClass().getDeclaredField(fieldName);
- field.setAccessible(true);
- return (T) field.get(instance);
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryOutputStreamSerializerTest.java b/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryOutputStreamSerializerTest.java
deleted file mode 100644
index c8f5b4a87ec..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryOutputStreamSerializerTest.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import org.apache.avro.Schema;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link GlueSchemaRegistryOutputStreamSerializer}. */
-class GlueSchemaRegistryOutputStreamSerializerTest {
- private static final String testTopic = "Test-Topic";
- private static final String AVRO_USER_SCHEMA_FILE = "src/test/java/resources/avro/user.avsc";
- private static final byte[] actualBytes =
- new byte[] {12, 99, 8, 116, 101, 115, 116, 0, 20, 0, 12, 118, 105, 111, 108, 101, 116};
- private static final byte[] specificBytes =
- new byte[] {
- 3, 0, -73, -76, -89, -16, -100, -106, 78, 74, -90, -121, -5, 93, -23, -17, 12, 99,
- 8, 116, 101, 115, 116, 0, 20, 0, 12, 118, 105, 111, 108, 101, 116
- };
- private static Schema userSchema;
- private static User userDefinedPojo;
- private static final Map<String, Object> configs = new HashMap<>();
- private static final Map<String, String> metadata = new HashMap<>();
- private static final AwsCredentialsProvider credentialsProvider =
- DefaultCredentialsProvider.builder().build();
- private static GlueSchemaRegistrySerializationFacade mockSerializationFacade;
-
- @BeforeAll
- static void setup() throws IOException {
- metadata.put("test-key", "test-value");
- metadata.put(AWSSchemaRegistryConstants.TRANSPORT_METADATA_KEY, testTopic);
-
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-
- Schema.Parser parser = new Schema.Parser();
- userSchema = parser.parse(new File(AVRO_USER_SCHEMA_FILE));
- userDefinedPojo =
- User.newBuilder()
- .setName("test_avro_schema")
- .setFavoriteColor("violet")
- .setFavoriteNumber(10)
- .build();
-
- mockSerializationFacade = new MockGlueSchemaRegistrySerializationFacade();
- }
-
- /**
- * Test whether constructor works with topic name and AWS Glue Schema Registry configuration
- * map.
- */
- @Test
- void testConstructor_withConfigsAndCredential_succeeds() {
- GlueSchemaRegistryOutputStreamSerializer glueSchemaRegistryOutputStreamSerializer =
- new GlueSchemaRegistryOutputStreamSerializer(testTopic, configs);
- assertThat(glueSchemaRegistryOutputStreamSerializer)
- .isInstanceOf(GlueSchemaRegistryOutputStreamSerializer.class);
- }
-
- /** Test whether constructor works with Glue Schema Registry SerializationFacade. */
- @Test
- void testConstructor_withDeserializer_succeeds() {
- GlueSchemaRegistryOutputStreamSerializer glueSchemaRegistryOutputStreamSerializer =
- new GlueSchemaRegistryOutputStreamSerializer(
- testTopic, configs, mockSerializationFacade);
- assertThat(glueSchemaRegistryOutputStreamSerializer)
- .isInstanceOf(GlueSchemaRegistryOutputStreamSerializer.class);
- }
-
- /** Test whether registerSchemaAndSerializeStream method works. */
- @Test
- void testRegisterSchemaAndSerializeStream_withValidParams_succeeds() throws IOException {
- GlueSchemaRegistryOutputStreamSerializer glueSchemaRegistryOutputStreamSerializer =
- new GlueSchemaRegistryOutputStreamSerializer(
- testTopic, configs, mockSerializationFacade);
- ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
- glueSchemaRegistryOutputStreamSerializer.registerSchemaAndSerializeStream(
- userSchema, outputStream, actualBytes);
-
- assertThat(outputStream.toByteArray()).isEqualTo(specificBytes);
- }
-
- private static class MockGlueSchemaRegistrySerializationFacade
- extends GlueSchemaRegistrySerializationFacade {
-
- public MockGlueSchemaRegistrySerializationFacade() {
- super(
- credentialsProvider,
- null,
- new GlueSchemaRegistryConfiguration(configs),
- configs,
- null);
- }
-
- @Override
- public byte[] encode(
- String transportName,
- com.amazonaws.services.schemaregistry.common.Schema schema,
- byte[] data) {
- return specificBytes;
- }
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/User.java b/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/User.java
deleted file mode 100644
index c16321d18e7..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/User.java
+++ /dev/null
@@ -1,434 +0,0 @@
-/*
- * 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.glue.schema.registry;
-
-import org.apache.avro.message.BinaryMessageDecoder;
-import org.apache.avro.message.BinaryMessageEncoder;
-import org.apache.avro.message.SchemaStore;
-import org.apache.avro.specific.SpecificData;
-
-@SuppressWarnings("all")
-@org.apache.avro.specific.AvroGenerated
-public class User extends org.apache.avro.specific.SpecificRecordBase
- implements org.apache.avro.specific.SpecificRecord {
- private static final long serialVersionUID = 2109752990030252627L;
- public static final org.apache.avro.Schema SCHEMA$ =
- new org.apache.avro.Schema.Parser()
- .parse(
- "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"com.amazonaws.services.schemaregistry.serializers.avro\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"favorite_number\",\"type\":[\"int\",\"null\"]},{\"name\":\"favorite_color\",\"type\":[\"string\",\"null\"]}]}");
-
- public static org.apache.avro.Schema getClassSchema() {
- return SCHEMA$;
- }
-
- private static SpecificData MODEL$ = new SpecificData();
-
- private static final BinaryMessageEncoder<User> ENCODER =
- new BinaryMessageEncoder<User>(MODEL$, SCHEMA$);
-
- private static final BinaryMessageDecoder<User> DECODER =
- new BinaryMessageDecoder<User>(MODEL$, SCHEMA$);
-
- /** Return the BinaryMessageDecoder instance used by this class. */
- public static BinaryMessageDecoder<User> getDecoder() {
- return DECODER;
- }
-
- /**
- * Create a new BinaryMessageDecoder instance for this class that uses the specified {@link
- * SchemaStore}.
- *
- * @param resolver a {@link SchemaStore} used to find schemas by fingerprint
- */
- public static BinaryMessageDecoder<User> createDecoder(SchemaStore resolver) {
- return new BinaryMessageDecoder<User>(MODEL$, SCHEMA$, resolver);
- }
-
- /** Serializes this User to a ByteBuffer. */
- public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
- return ENCODER.encode(this);
- }
-
- /** Deserializes a User from a ByteBuffer. */
- public static User fromByteBuffer(java.nio.ByteBuffer b) throws java.io.IOException {
- return DECODER.decode(b);
- }
-
- @Deprecated public CharSequence name;
- @Deprecated public Integer favorite_number;
- @Deprecated public CharSequence favorite_color;
-
- /**
- * Default constructor. Note that this does not initialize fields to their default values from
- * the schema. If that is desired then one should use <code>newBuilder()</code>.
- */
- public User() {}
-
- /**
- * All-args constructor.
- *
- * @param name The new value for name
- * @param favorite_number The new value for favorite_number
- * @param favorite_color The new value for favorite_color
- */
- public User(CharSequence name, Integer favorite_number, CharSequence favorite_color) {
- this.name = name;
- this.favorite_number = favorite_number;
- this.favorite_color = favorite_color;
- }
-
- public org.apache.avro.Schema getSchema() {
- return SCHEMA$;
- }
- // Used by DatumWriter. Applications should not call.
- public Object get(int field$) {
- switch (field$) {
- case 0:
- return name;
- case 1:
- return favorite_number;
- case 2:
- return favorite_color;
- default:
- throw new org.apache.avro.AvroRuntimeException("Bad index");
- }
- }
-
- // Used by DatumReader. Applications should not call.
- @SuppressWarnings(value = "unchecked")
- public void put(int field$, Object value$) {
- switch (field$) {
- case 0:
- name = (CharSequence) value$;
- break;
- case 1:
- favorite_number = (Integer) value$;
- break;
- case 2:
- favorite_color = (CharSequence) value$;
- break;
- default:
- throw new org.apache.avro.AvroRuntimeException("Bad index");
- }
- }
-
- /**
- * Gets the value of the 'name' field.
- *
- * @return The value of the 'name' field.
- */
- public CharSequence getName() {
- return name;
- }
-
- /**
- * Sets the value of the 'name' field.
- *
- * @param value the value to set.
- */
- public void setName(CharSequence value) {
- this.name = value;
- }
-
- /**
- * Gets the value of the 'favorite_number' field.
- *
- * @return The value of the 'favorite_number' field.
- */
- public Integer getFavoriteNumber() {
- return favorite_number;
- }
-
- /**
- * Sets the value of the 'favorite_number' field.
- *
- * @param value the value to set.
- */
- public void setFavoriteNumber(Integer value) {
- this.favorite_number = value;
- }
-
- /**
- * Gets the value of the 'favorite_color' field.
- *
- * @return The value of the 'favorite_color' field.
- */
- public CharSequence getFavoriteColor() {
- return favorite_color;
- }
-
- /**
- * Sets the value of the 'favorite_color' field.
- *
- * @param value the value to set.
- */
- public void setFavoriteColor(CharSequence value) {
- this.favorite_color = value;
- }
-
- /**
- * Creates a new User RecordBuilder.
- *
- * @return A new User RecordBuilder
- */
- public static Builder newBuilder() {
- return new Builder();
- }
-
- /**
- * Creates a new User RecordBuilder by copying an existing Builder.
- *
- * @param other The existing builder to copy.
- * @return A new User RecordBuilder
- */
- public static Builder newBuilder(Builder other) {
- return new Builder(other);
- }
-
- /**
- * Creates a new User RecordBuilder by copying an existing User instance.
- *
- * @param other The existing instance to copy.
- * @return A new User RecordBuilder
- */
- public static Builder newBuilder(User other) {
- return new Builder(other);
- }
-
- /** RecordBuilder for User instances. */
- public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<User>
- implements org.apache.avro.data.RecordBuilder<User> {
-
- private CharSequence name;
- private Integer favorite_number;
- private CharSequence favorite_color;
-
- /** Creates a new Builder */
- private Builder() {
- super(SCHEMA$);
- }
-
- /**
- * Creates a Builder by copying an existing Builder.
- *
- * @param other The existing Builder to copy.
- */
- private Builder(Builder other) {
- super(other);
- if (isValidValue(fields()[0], other.name)) {
- this.name = data().deepCopy(fields()[0].schema(), other.name);
- fieldSetFlags()[0] = true;
- }
- if (isValidValue(fields()[1], other.favorite_number)) {
- this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number);
- fieldSetFlags()[1] = true;
- }
- if (isValidValue(fields()[2], other.favorite_color)) {
- this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color);
- fieldSetFlags()[2] = true;
- }
- }
-
- /**
- * Creates a Builder by copying an existing User instance
- *
- * @param other The existing instance to copy.
- */
- private Builder(User other) {
- super(SCHEMA$);
- if (isValidValue(fields()[0], other.name)) {
- this.name = data().deepCopy(fields()[0].schema(), other.name);
- fieldSetFlags()[0] = true;
- }
- if (isValidValue(fields()[1], other.favorite_number)) {
- this.favorite_number = data().deepCopy(fields()[1].schema(), other.favorite_number);
- fieldSetFlags()[1] = true;
- }
- if (isValidValue(fields()[2], other.favorite_color)) {
- this.favorite_color = data().deepCopy(fields()[2].schema(), other.favorite_color);
- fieldSetFlags()[2] = true;
- }
- }
-
- /**
- * Gets the value of the 'name' field.
- *
- * @return The value.
- */
- public CharSequence getName() {
- return name;
- }
-
- /**
- * Sets the value of the 'name' field.
- *
- * @param value The value of 'name'.
- * @return This builder.
- */
- public Builder setName(CharSequence value) {
- validate(fields()[0], value);
- this.name = value;
- fieldSetFlags()[0] = true;
- return this;
- }
-
- /**
- * Checks whether the 'name' field has been set.
- *
- * @return True if the 'name' field has been set, false otherwise.
- */
- public boolean hasName() {
- return fieldSetFlags()[0];
- }
-
- /**
- * Clears the value of the 'name' field.
- *
- * @return This builder.
- */
- public Builder clearName() {
- name = null;
- fieldSetFlags()[0] = false;
- return this;
- }
-
- /**
- * Gets the value of the 'favorite_number' field.
- *
- * @return The value.
- */
- public Integer getFavoriteNumber() {
- return favorite_number;
- }
-
- /**
- * Sets the value of the 'favorite_number' field.
- *
- * @param value The value of 'favorite_number'.
- * @return This builder.
- */
- public Builder setFavoriteNumber(Integer value) {
- validate(fields()[1], value);
- this.favorite_number = value;
- fieldSetFlags()[1] = true;
- return this;
- }
-
- /**
- * Checks whether the 'favorite_number' field has been set.
- *
- * @return True if the 'favorite_number' field has been set, false otherwise.
- */
- public boolean hasFavoriteNumber() {
- return fieldSetFlags()[1];
- }
-
- /**
- * Clears the value of the 'favorite_number' field.
- *
- * @return This builder.
- */
- public Builder clearFavoriteNumber() {
- favorite_number = null;
- fieldSetFlags()[1] = false;
- return this;
- }
-
- /**
- * Gets the value of the 'favorite_color' field.
- *
- * @return The value.
- */
- public CharSequence getFavoriteColor() {
- return favorite_color;
- }
-
- /**
- * Sets the value of the 'favorite_color' field.
- *
- * @param value The value of 'favorite_color'.
- * @return This builder.
- */
- public Builder setFavoriteColor(CharSequence value) {
- validate(fields()[2], value);
- this.favorite_color = value;
- fieldSetFlags()[2] = true;
- return this;
- }
-
- /**
- * Checks whether the 'favorite_color' field has been set.
- *
- * @return True if the 'favorite_color' field has been set, false otherwise.
- */
- public boolean hasFavoriteColor() {
- return fieldSetFlags()[2];
- }
-
- /**
- * Clears the value of the 'favorite_color' field.
- *
- * @return This builder.
- */
- public Builder clearFavoriteColor() {
- favorite_color = null;
- fieldSetFlags()[2] = false;
- return this;
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public User build() {
- try {
- User record = new User();
- record.name =
- fieldSetFlags()[0] ? this.name : (CharSequence) defaultValue(fields()[0]);
- record.favorite_number =
- fieldSetFlags()[1]
- ? this.favorite_number
- : (Integer) defaultValue(fields()[1]);
- record.favorite_color =
- fieldSetFlags()[2]
- ? this.favorite_color
- : (CharSequence) defaultValue(fields()[2]);
- return record;
- } catch (Exception e) {
- throw new org.apache.avro.AvroRuntimeException(e);
- }
- }
- }
-
- @SuppressWarnings("unchecked")
- private static final org.apache.avro.io.DatumWriter<User> WRITER$ =
- (org.apache.avro.io.DatumWriter<User>) MODEL$.createDatumWriter(SCHEMA$);
-
- @Override
- public void writeExternal(java.io.ObjectOutput out) throws java.io.IOException {
- WRITER$.write(this, SpecificData.getEncoder(out));
- }
-
- @SuppressWarnings("unchecked")
- private static final org.apache.avro.io.DatumReader<User> READER$ =
- (org.apache.avro.io.DatumReader<User>) MODEL$.createDatumReader(SCHEMA$);
-
- @Override
- public void readExternal(java.io.ObjectInput in) throws java.io.IOException {
- READER$.read(this, SpecificData.getDecoder(in));
- }
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/java/resources/avro/user.avsc b/flink-formats/flink-avro-glue-schema-registry/src/test/java/resources/avro/user.avsc
deleted file mode 100644
index a13e075903b..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/java/resources/avro/user.avsc
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-{
- "namespace": "org.apache.flink.formats.avro.glue.schema.registry",
- "type": "record",
- "name": "User",
- "fields": [
- {"name": "name", "type": "string" },
- {"name": "favorite_number", "type": ["int", "null"] },
- {"name": "favorite_color", "type": ["string", "null"] }
- ]
-}
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/flink-formats/flink-avro-glue-schema-registry/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
deleted file mode 100644
index 28999133c2b..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-org.apache.flink.util.TestLoggerExtension
\ No newline at end of file
diff --git a/flink-formats/flink-avro-glue-schema-registry/src/test/resources/archunit.properties b/flink-formats/flink-avro-glue-schema-registry/src/test/resources/archunit.properties
deleted file mode 100644
index 15be88c95ba..00000000000
--- a/flink-formats/flink-avro-glue-schema-registry/src/test/resources/archunit.properties
+++ /dev/null
@@ -1,31 +0,0 @@
-#
-# 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.
-#
-
-# By default we allow removing existing violations, but fail when new violations are added.
-freeze.store.default.allowStoreUpdate=true
-
-# Enable this if a new (frozen) rule has been added in order to create the initial store and record the existing violations.
-#freeze.store.default.allowStoreCreation=true
-
-# Enable this to add allow new violations to be recorded.
-# NOTE: Adding new violations should be avoided when possible. If the rule was correct to flag a new
-# violation, please try to avoid creating the violation. If the violation was created due to a
-# shortcoming of the rule, file a JIRA issue so the rule can be improved.
-#freeze.refreeze=true
-
-freeze.store.default.path=archunit-violations
diff --git a/flink-formats/flink-json-glue-schema-registry/archunit-violations/4703059b-4f06-41c9-9724-644e6d00584f b/flink-formats/flink-json-glue-schema-registry/archunit-violations/4703059b-4f06-41c9-9724-644e6d00584f
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/flink-formats/flink-json-glue-schema-registry/archunit-violations/b99819a4-a946-475e-883f-963de77c7e57 b/flink-formats/flink-json-glue-schema-registry/archunit-violations/b99819a4-a946-475e-883f-963de77c7e57
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/flink-formats/flink-json-glue-schema-registry/archunit-violations/stored.rules b/flink-formats/flink-json-glue-schema-registry/archunit-violations/stored.rules
deleted file mode 100644
index 463e6804b24..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/archunit-violations/stored.rules
+++ /dev/null
@@ -1,4 +0,0 @@
-#
-#Mon Apr 04 17:19:25 CEST 2022
-Tests\ inheriting\ from\ AbstractTestBase\ should\ have\ name\ ending\ with\ ITCase=4703059b-4f06-41c9-9724-644e6d00584f
-ITCASE\ tests\ should\ use\ a\ MiniCluster\ resource\ or\ extension=b99819a4-a946-475e-883f-963de77c7e57
diff --git a/flink-formats/flink-json-glue-schema-registry/pom.xml b/flink-formats/flink-json-glue-schema-registry/pom.xml
deleted file mode 100644
index 0b4dbebcddf..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/pom.xml
+++ /dev/null
@@ -1,156 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <parent>
- <artifactId>flink-formats</artifactId>
- <groupId>org.apache.flink</groupId>
- <version>1.17-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
-
- <artifactId>flink-json-glue-schema-registry_${scala.binary.version}</artifactId>
- <name>Flink : Formats : JSON AWS Glue Schema Registry</name>
- <packaging>jar</packaging>
-
- <properties>
- <glue.schema.registry.version>1.1.14</glue.schema.registry.version>
- <aws.sdkv2.version>2.19.14</aws.sdkv2.version>
- <netty.version>4.1.68.Final</netty.version>
- <kotlin.version>1.7.10</kotlin.version>
- <classgraph.version>4.8.120</classgraph.version>
- </properties>
-
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>software.amazon.awssdk</groupId>
- <artifactId>bom</artifactId>
- <version>${aws.sdkv2.version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
-
- <!-- For dependency convergence -->
- <dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- <version>3.21.7</version>
- </dependency>
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>30.0-jre</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-reflect</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-common</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-stdlib-jdk8</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlin</groupId>
- <artifactId>kotlin-scripting-compiler-embeddable</artifactId>
- <version>${kotlin.version}</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains.kotlinx</groupId>
- <artifactId>kotlinx-serialization-core-jvm</artifactId>
- <version>1.0.1</version>
- </dependency>
- <dependency>
- <groupId>io.github.classgraph</groupId>
- <artifactId>classgraph</artifactId>
- <version>4.8.120</version>
- </dependency>
- <dependency>
- <groupId>org.jetbrains</groupId>
- <artifactId>annotations</artifactId>
- <version>17.0.0</version>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
- <dependencies>
- <!-- core dependencies -->
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-core</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-connector-aws-base</artifactId>
- <version>${project.version}</version>
- </dependency>
-
- <dependency>
- <groupId>software.amazon.glue</groupId>
- <artifactId>schema-registry-serde</artifactId>
- <version>${glue.schema.registry.version}</version>
- </dependency>
-
- <dependency>
- <groupId>software.amazon.glue</groupId>
- <artifactId>schema-registry-common</artifactId>
- <version>${glue.schema.registry.version}</version>
- </dependency>
-
- <dependency>
- <groupId>com.kjetland</groupId>
- <artifactId>mbknor-jackson-jsonschema_${scala.binary.version}</artifactId>
- <version>1.0.39</version>
- </dependency>
-
- <!-- ArchUit test dependencies -->
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-architecture-tests-test</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
-</project>
diff --git a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonDeserializationSchema.java b/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonDeserializationSchema.java
deleted file mode 100644
index d7cc46b7a2c..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonDeserializationSchema.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.serialization.DeserializationSchema;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-
-import java.util.Map;
-
-/**
- * AWS Glue Schema Registry Deserialization schema to de-serialize JSON Schema binary format for
- * Flink Consumer user.
- *
- * @param <T> type of record it produces
- */
-@PublicEvolving
-public class GlueSchemaRegistryJsonDeserializationSchema<T> implements DeserializationSchema<T> {
-
- private final Class<T> recordClazz;
- private final GlueSchemaRegistryJsonSchemaCoderProvider
- glueSchemaRegistryJsonSchemaCoderProvider;
- protected GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder;
-
- /**
- * Creates a JSON Schema deserialization schema.
- *
- * @param recordClazz class to which deserialize. Should be one of: {@link
- * com.amazonaws.services.schemaregistry.serializers.json.JsonDataWithSchema}, or user
- * defined POJO.
- * @param transportName topic name or stream name etc.
- * @param configs configuration map of AWS Glue Schema Registry
- */
- public GlueSchemaRegistryJsonDeserializationSchema(
- Class<T> recordClazz, String transportName, Map<String, Object> configs) {
- this.recordClazz = recordClazz;
- this.glueSchemaRegistryJsonSchemaCoderProvider =
- new GlueSchemaRegistryJsonSchemaCoderProvider(transportName, configs);
- }
-
- @VisibleForTesting
- protected GlueSchemaRegistryJsonDeserializationSchema(
- Class<T> recordClazz,
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder) {
- this.recordClazz = recordClazz;
- this.glueSchemaRegistryJsonSchemaCoderProvider = null;
- this.glueSchemaRegistryJsonSchemaCoder = glueSchemaRegistryJsonSchemaCoder;
- }
-
- /**
- * Deserializes the incoming byte array which contains bytes of AWS Glue Schema registry
- * information back to the original object.
- *
- * @param bytes The incoming byte array to be deserialized
- * @return The deserialized object.
- */
- @Override
- public T deserialize(byte[] bytes) {
- if (bytes == null) {
- return null;
- }
-
- if (glueSchemaRegistryJsonSchemaCoder == null) {
- glueSchemaRegistryJsonSchemaCoder = glueSchemaRegistryJsonSchemaCoderProvider.get();
- }
- return (T) glueSchemaRegistryJsonSchemaCoder.deserialize(bytes);
- }
-
- @Override
- public boolean isEndOfStream(T t) {
- return false;
- }
-
- @Override
- public TypeInformation<T> getProducedType() {
- return TypeInformation.of(recordClazz);
- }
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoder.java b/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoder.java
deleted file mode 100644
index c02f8ce4367..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoder.java
+++ /dev/null
@@ -1,119 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.connector.aws.util.AWSGeneralUtil;
-
-import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
-import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.utils.GlueSchemaRegistryUtils;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.io.Serializable;
-import java.nio.ByteBuffer;
-import java.util.Map;
-import java.util.UUID;
-
-/**
- * Schema coder that allows reading schema that is somehow embedded into serialized record. Used by
- * {@link GlueSchemaRegistryJsonSerializationSchema} and {@link
- * GlueSchemaRegistryJsonSerializationSchema}.
- */
-@PublicEvolving
-public class GlueSchemaRegistryJsonSchemaCoder implements Serializable {
-
- private final String transportName;
- private final Map<String, Object> configs;
- private final GlueSchemaRegistrySerializationFacade glueSchemaRegistrySerializationFacade;
- private final GlueSchemaRegistryDeserializationFacade glueSchemaRegistryDeserializationFacade;
-
- /**
- * Constructor accepts transport name and configuration map for AWS Glue Schema Registry.
- *
- * @param transportName topic name or stream name etc.
- * @param configs configurations for AWS Glue Schema Registry
- */
- public GlueSchemaRegistryJsonSchemaCoder(
- final String transportName, final Map<String, Object> configs) {
- this.transportName = transportName;
- this.configs = configs;
-
- AwsCredentialsProvider credentialsProvider = AWSGeneralUtil.getCredentialsProvider(configs);
-
- this.glueSchemaRegistrySerializationFacade =
- GlueSchemaRegistrySerializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .glueSchemaRegistryConfiguration(
- new GlueSchemaRegistryConfiguration(configs))
- .build();
- this.glueSchemaRegistryDeserializationFacade =
- GlueSchemaRegistryDeserializationFacade.builder()
- .credentialProvider(credentialsProvider)
- .configs(configs)
- .build();
- }
-
- @VisibleForTesting
- protected GlueSchemaRegistryJsonSchemaCoder(
- final String transportName,
- final Map<String, Object> configs,
- GlueSchemaRegistrySerializationFacade glueSchemaRegistrySerializationFacade,
- GlueSchemaRegistryDeserializationFacade glueSchemaRegistryDeserializationFacade) {
- this.transportName = transportName;
- this.configs = configs;
- this.glueSchemaRegistrySerializationFacade = glueSchemaRegistrySerializationFacade;
- this.glueSchemaRegistryDeserializationFacade = glueSchemaRegistryDeserializationFacade;
- }
-
- public Object deserialize(byte[] bytes) {
- return glueSchemaRegistryDeserializationFacade.deserialize(
- AWSDeserializerInput.builder()
- .transportName(transportName)
- .buffer(ByteBuffer.wrap(bytes))
- .build());
- }
-
- public byte[] registerSchemaAndSerialize(Object object) {
- String schema =
- glueSchemaRegistrySerializationFacade.getSchemaDefinition(DataFormat.JSON, object);
- String schemaName = GlueSchemaRegistryUtils.getInstance().getSchemaName(configs);
- schemaName =
- schemaName != null
- ? schemaName
- : GlueSchemaRegistryUtils.getInstance()
- .configureSchemaNamingStrategy(configs)
- .getSchemaName(transportName);
- UUID schemaVersionId =
- glueSchemaRegistrySerializationFacade.getOrRegisterSchemaVersion(
- AWSSerializerInput.builder()
- .dataFormat(DataFormat.JSON.name())
- .schemaDefinition(schema)
- .schemaName(schemaName)
- .transportName(transportName)
- .build());
- return glueSchemaRegistrySerializationFacade.serialize(
- DataFormat.JSON, object, schemaVersionId);
- }
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoderProvider.java b/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoderProvider.java
deleted file mode 100644
index bb9902b14d9..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoderProvider.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-
-import lombok.NonNull;
-
-import java.io.Serializable;
-import java.util.Map;
-
-/** Provider for {@link GlueSchemaRegistryJsonSchemaCoder}. */
-@PublicEvolving
-public class GlueSchemaRegistryJsonSchemaCoderProvider implements Serializable {
-
- private final String transportName;
- private final Map<String, Object> configs;
-
- /**
- * Constructor used by {@link GlueSchemaRegistryJsonSerializationSchema} and {@link
- * GlueSchemaRegistryJsonDeserializationSchema}.
- *
- * @param transportName topic name or stream name etc.
- * @param configs configurations for AWS Glue Schema Registry
- */
- public GlueSchemaRegistryJsonSchemaCoderProvider(
- String transportName, @NonNull Map<String, Object> configs) {
- this.transportName = transportName;
- this.configs = configs;
- }
-
- public GlueSchemaRegistryJsonSchemaCoder get() {
- return new GlueSchemaRegistryJsonSchemaCoder(transportName, configs);
- }
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSerializationSchema.java b/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSerializationSchema.java
deleted file mode 100644
index 98dfaa4f083..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/main/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSerializationSchema.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.api.common.serialization.SerializationSchema;
-
-import java.util.Map;
-
-/**
- * AWS Glue Schema Registry Serialization schema to serialize to JSON Schema binary format for Flink
- * Producer user.
- *
- * @param <T> the type to be serialized
- */
-@PublicEvolving
-public class GlueSchemaRegistryJsonSerializationSchema<T> implements SerializationSchema<T> {
-
- private final GlueSchemaRegistryJsonSchemaCoderProvider
- glueSchemaRegistryJsonSchemaCoderProvider;
- protected GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder;
-
- /**
- * Creates a JSON Schema serialization schema.
- *
- * @param transportName topic name or stream name etc.
- * @param configs configuration map of AWS Glue Schema Registry
- */
- public GlueSchemaRegistryJsonSerializationSchema(
- String transportName, Map<String, Object> configs) {
- this.glueSchemaRegistryJsonSchemaCoderProvider =
- new GlueSchemaRegistryJsonSchemaCoderProvider(transportName, configs);
- }
-
- @VisibleForTesting
- protected GlueSchemaRegistryJsonSerializationSchema(
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder) {
- this.glueSchemaRegistryJsonSchemaCoderProvider = null;
- this.glueSchemaRegistryJsonSchemaCoder = glueSchemaRegistryJsonSchemaCoder;
- }
-
- /**
- * Serializes the incoming element to a byte array containing bytes of AWS Glue Schema registry
- * information.
- *
- * @param object The incoming element to be serialized
- * @return The serialized bytes.
- */
- @Override
- public byte[] serialize(T object) {
- if (object == null) {
- return null;
- }
-
- if (glueSchemaRegistryJsonSchemaCoder == null) {
- glueSchemaRegistryJsonSchemaCoder = glueSchemaRegistryJsonSchemaCoderProvider.get();
- }
- return glueSchemaRegistryJsonSchemaCoder.registerSchemaAndSerialize(object);
- }
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/architecture/TestCodeArchitectureTest.java b/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/architecture/TestCodeArchitectureTest.java
deleted file mode 100644
index 0724f1582fa..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/architecture/TestCodeArchitectureTest.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- * 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.architecture;
-
-import org.apache.flink.architecture.common.ImportOptions;
-
-import com.tngtech.archunit.core.importer.ImportOption;
-import com.tngtech.archunit.junit.AnalyzeClasses;
-import com.tngtech.archunit.junit.ArchTest;
-import com.tngtech.archunit.junit.ArchTests;
-
-/** Architecture tests for test code. */
-@AnalyzeClasses(
- packages = {"org.apache.flink.formats.json.glue.schema.registry"},
- importOptions = {
- ImportOption.OnlyIncludeTests.class,
- ImportOptions.ExcludeScalaImportOption.class,
- ImportOptions.ExcludeShadedImportOption.class
- })
-public class TestCodeArchitectureTest {
-
- @ArchTest
- public static final ArchTests COMMON_TESTS = ArchTests.in(TestCodeArchitectureTestBase.class);
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/Car.java b/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/Car.java
deleted file mode 100644
index e3966470f7c..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/Car.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import com.fasterxml.jackson.annotation.JsonFormat;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaDefault;
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaDescription;
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject;
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInt;
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaString;
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.EqualsAndHashCode;
-
-import javax.validation.constraints.Max;
-import javax.validation.constraints.Min;
-
-import java.util.Collection;
-import java.util.Date;
-
-/** POJO defined JSON Schema. */
-@JsonSchemaDescription("This is a car")
-@JsonSchemaTitle("Simple Car Schema")
-@Builder
-@AllArgsConstructor
-@EqualsAndHashCode
-// Fully qualified class name to be added to an additionally injected property
-// called className for deserializer to determine which class to deserialize
-// the bytes into
-@JsonSchemaInject(
- strings = {
- @JsonSchemaString(
- path = "className",
- value = "org.apache.flink.formats.jsonschema.glue.schema.registry.Car")
- })
-// List of annotations to help infer JSON Schema are defined by
-// https://github.com/mbknor/mbknor-jackson-jsonSchema
-public class Car {
- @JsonProperty(required = true)
- private String make;
-
- @JsonProperty(required = true)
- private String model;
-
- @JsonSchemaDefault("true")
- @JsonProperty
- public boolean used;
-
- @JsonSchemaInject(ints = {@JsonSchemaInt(path = "multipleOf", value = 1000)})
- @Max(200000)
- @JsonProperty
- private int miles;
-
- @Min(2000)
- @JsonProperty
- private int year;
-
- @JsonProperty private Date purchaseDate;
-
- @JsonProperty
- @JsonFormat(shape = JsonFormat.Shape.NUMBER)
- private Date listedDate;
-
- @JsonProperty private String[] owners;
-
- @JsonProperty private Collection<Float> serviceChecks;
-
- // Empty constructor is required by Jackson to deserialize bytes
- // into an Object of this class
- public Car() {}
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonDeserializationSchemaTest.java b/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonDeserializationSchemaTest.java
deleted file mode 100644
index 83c09eb386f..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonDeserializationSchemaTest.java
+++ /dev/null
@@ -1,182 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import com.amazonaws.services.schemaregistry.common.AWSDeserializerInput;
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.exception.AWSSchemaRegistryException;
-import com.amazonaws.services.schemaregistry.serializers.json.JsonDataWithSchema;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import lombok.NonNull;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-
-import java.time.Instant;
-import java.util.Arrays;
-import java.util.Calendar;
-import java.util.Date;
-import java.util.GregorianCalendar;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link GlueSchemaRegistryJsonDeserializationSchema}. */
-class GlueSchemaRegistryJsonDeserializationSchemaTest {
- private static final String testTopic = "Test-Topic";
- private static final Map<String, Object> configs = new HashMap<>();
- private static final AwsCredentialsProvider credentialsProvider =
- DefaultCredentialsProvider.builder().build();
- private static final byte[] serializedBytes =
- new byte[] {
- 3, 0, -73, -76, -89, -16, -100, -106, 78, 74, -90, -121, -5, 93, -23, -17, 12, 99,
- 8, 116, 101, 115, 116, 0, 20, 0, 12, 118, 105, 111, 108, 101, 116
- };
- private static final String JSON_SCHEMA =
- "{\n"
- + " \"$id\": \"https://example.com/address.schema.json\",\n"
- + " \"$schema\": \"http://json-schema.org/draft-07/schema#\",\n"
- + " \"type\": \"object\",\n"
- + " \"properties\": {\n"
- + " \"f1\": { \"type\": \"string\" },\n"
- + " \"f2\": { \"type\": \"integer\", \"maximum\": 1000 }\n"
- + " }\n"
- + " }";
- private static JsonDataWithSchema userSchema;
- private static Car userDefinedPojo;
- private static GlueSchemaRegistryDeserializationFacade mockDeserializationFacadeForSpecific;
- private static GlueSchemaRegistryDeserializationFacade mockDeserializationFacadeForGeneric;
-
- @BeforeAll
- static void setup() {
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
-
- userSchema =
- JsonDataWithSchema.builder(JSON_SCHEMA, "{\"f1\":\"iphone\",\"f2\":12}").build();
- userDefinedPojo =
- Car.builder()
- .make("Tesla")
- .model("3")
- .used(false)
- .miles(6000)
- .year(2021)
- .listedDate(new GregorianCalendar(2020, Calendar.FEBRUARY, 20).getTime())
- .purchaseDate(Date.from(Instant.parse("2020-01-01T00:00:00.000Z")))
- .owners(new String[] {"Harry", "Megan"})
- .serviceChecks(Arrays.asList(5000.0f, 10780.30f))
- .build();
-
- mockDeserializationFacadeForSpecific =
- new MockGlueSchemaRegistrySerializationFacadeForSpecific();
- mockDeserializationFacadeForGeneric =
- new MockGlueSchemaRegistrySerializationFacadeForGeneric();
- }
-
- /** Test initialization for generic type JSON Schema works. */
- @Test
- void testForGeneric_withValidParams_succeeds() {
- assertThat(
- new GlueSchemaRegistryJsonDeserializationSchema<>(
- JsonDataWithSchema.class, testTopic, configs))
- .isNotNull();
- }
-
- /** Test initialization for specific type JSON Schema works. */
- @Test
- void testForSpecific_withValidParams_succeeds() {
- assertThat(new GlueSchemaRegistryJsonDeserializationSchema<>(Car.class, testTopic, configs))
- .isNotNull();
- }
-
- /** Test whether deserialize method for specific type JSON Schema data works. */
- @Test
- void testDeserializePOJO_withValidParams_succeeds() {
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder =
- new GlueSchemaRegistryJsonSchemaCoder(
- testTopic, configs, null, mockDeserializationFacadeForSpecific);
-
- GlueSchemaRegistryJsonDeserializationSchema<Car>
- glueSchemaRegistryJsonDeserializationSchema =
- new GlueSchemaRegistryJsonDeserializationSchema<>(
- Car.class, glueSchemaRegistryJsonSchemaCoder);
-
- Object deserializedObject =
- glueSchemaRegistryJsonDeserializationSchema.deserialize(serializedBytes);
- assertThat(deserializedObject).isInstanceOf(Car.class).isEqualTo(userDefinedPojo);
- }
-
- /** Test whether deserialize method for generic type JSON Schema data works. */
- @Test
- void testDeserializeGenericData_withValidParams_succeeds() {
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder =
- new GlueSchemaRegistryJsonSchemaCoder(
- testTopic, configs, null, mockDeserializationFacadeForGeneric);
-
- GlueSchemaRegistryJsonDeserializationSchema<Car>
- glueSchemaRegistryJsonDeserializationSchema =
- new GlueSchemaRegistryJsonDeserializationSchema<>(
- Car.class, glueSchemaRegistryJsonSchemaCoder);
-
- Object deserializedObject =
- glueSchemaRegistryJsonDeserializationSchema.deserialize(serializedBytes);
- assertThat(deserializedObject).isInstanceOf(JsonDataWithSchema.class).isEqualTo(userSchema);
- }
-
- /** Test whether deserialize method returns null when input byte array is null. */
- @Test
- void testDeserialize_withNullObject_returnNull() {
- GlueSchemaRegistryJsonDeserializationSchema<Car>
- glueSchemaRegistryJsonDeserializationSchema =
- new GlueSchemaRegistryJsonDeserializationSchema<>(
- Car.class, testTopic, configs);
- assertThat(glueSchemaRegistryJsonDeserializationSchema.deserialize(null)).isNull();
- }
-
- private static class MockGlueSchemaRegistrySerializationFacadeForSpecific
- extends GlueSchemaRegistryDeserializationFacade {
-
- public MockGlueSchemaRegistrySerializationFacadeForSpecific() {
- super(configs, null, credentialsProvider, null);
- }
-
- @Override
- public Object deserialize(@NonNull AWSDeserializerInput deserializerInput)
- throws AWSSchemaRegistryException {
- return userDefinedPojo;
- }
- }
-
- private static class MockGlueSchemaRegistrySerializationFacadeForGeneric
- extends GlueSchemaRegistryDeserializationFacade {
-
- public MockGlueSchemaRegistrySerializationFacadeForGeneric() {
- super(configs, null, credentialsProvider, null);
- }
-
- @Override
- public Object deserialize(@NonNull AWSDeserializerInput deserializerInput)
- throws AWSSchemaRegistryException {
- return userSchema;
- }
- }
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoderTest.java b/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoderTest.java
deleted file mode 100644
index af57af577d3..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSchemaCoderTest.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import com.amazonaws.services.schemaregistry.deserializers.GlueSchemaRegistryDeserializationFacade;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-
-import java.lang.reflect.Field;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_ACCESS_KEY_ID;
-import static org.apache.flink.connector.aws.config.AWSConfigConstants.AWS_SECRET_ACCESS_KEY;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link GlueSchemaRegistryJsonSchemaCoder}. */
-class GlueSchemaRegistryJsonSchemaCoderTest {
-
- @Test
- void testDefaultAwsCredentialsProvider() throws Exception {
- GlueSchemaRegistryJsonSchemaCoder coder =
- new GlueSchemaRegistryJsonSchemaCoder("test", getBaseConfig());
-
- GlueSchemaRegistryDeserializationFacade facade =
- getField("glueSchemaRegistryDeserializationFacade", coder);
-
- AwsCredentialsProvider credentialsProvider = facade.getCredentialsProvider();
- assertThat(credentialsProvider).isInstanceOf(DefaultCredentialsProvider.class);
- }
-
- @Test
- void testAwsCredentialsProviderFromConfig() throws Exception {
- Map<String, Object> config = new HashMap<>(getBaseConfig());
- config.put(AWS_ACCESS_KEY_ID, "ak");
- config.put(AWS_SECRET_ACCESS_KEY, "sk");
-
- GlueSchemaRegistryJsonSchemaCoder coder =
- new GlueSchemaRegistryJsonSchemaCoder("test", config);
-
- GlueSchemaRegistryDeserializationFacade facade =
- getField("glueSchemaRegistryDeserializationFacade", coder);
-
- AwsCredentialsProvider credentialsProvider = facade.getCredentialsProvider();
- assertThat(credentialsProvider.resolveCredentials().accessKeyId()).isEqualTo("ak");
- assertThat(credentialsProvider.resolveCredentials().secretAccessKey()).isEqualTo("sk");
- }
-
- private <T> T getField(final String fieldName, final Object instance) throws Exception {
- Field field = instance.getClass().getDeclaredField(fieldName);
- field.setAccessible(true);
- return (T) field.get(instance);
- }
-
- private Map<String, Object> getBaseConfig() {
- Map<String, Object> configs = new HashMap<>();
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
- return configs;
- }
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSerializationSchemaTest.java b/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSerializationSchemaTest.java
deleted file mode 100644
index ef80dcdda76..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/test/java/org/apache/flink/formats/json/glue/schema/registry/GlueSchemaRegistryJsonSerializationSchemaTest.java
+++ /dev/null
@@ -1,224 +0,0 @@
-/*
- * 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.json.glue.schema.registry;
-
-import com.amazonaws.services.schemaregistry.common.AWSSerializerInput;
-import com.amazonaws.services.schemaregistry.common.configs.GlueSchemaRegistryConfiguration;
-import com.amazonaws.services.schemaregistry.serializers.GlueSchemaRegistrySerializationFacade;
-import com.amazonaws.services.schemaregistry.serializers.json.JsonDataWithSchema;
-import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
-import lombok.NonNull;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-import software.amazon.awssdk.services.glue.model.DataFormat;
-
-import java.time.Instant;
-import java.util.Arrays;
-import java.util.Calendar;
-import java.util.Date;
-import java.util.GregorianCalendar;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
-import static com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants.COMPRESSION.NONE;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Tests for {@link GlueSchemaRegistryJsonSerializationSchema}. */
-class GlueSchemaRegistryJsonSerializationSchemaTest {
- private static final String testTopic = "Test-Topic";
- private static final String schemaName = "User-Topic";
- private static final String JSON_SCHEMA =
- "{\n"
- + " \"$id\": \"https://example.com/address.schema.json\",\n"
- + " \"$schema\": \"http://json-schema.org/draft-07/schema#\",\n"
- + " \"type\": \"object\",\n"
- + " \"properties\": {\n"
- + " \"f1\": { \"type\": \"string\" },\n"
- + " \"f2\": { \"type\": \"integer\", \"maximum\": 1000 }\n"
- + " }\n"
- + " }";
- private static final byte[] serializedBytes =
- new byte[] {
- 3, 0, -73, -76, -89, -16, -100, -106, 78, 74, -90, -121, -5, 93, -23, -17, 12, 99,
- 8, 116, 101, 115, 116, 0, 20, 0, 12, 118, 105, 111, 108, 101, 116
- };
- private static final UUID schemaVersionId = UUID.randomUUID();
- private static JsonDataWithSchema userSchema;
- private static Car userDefinedPojo;
- private static final Map<String, Object> configs = new HashMap<>();
- private static final Map<String, String> metadata = new HashMap<>();
- private static GlueSchemaRegistryConfiguration glueSchemaRegistryConfiguration;
- private static final AwsCredentialsProvider credentialsProvider =
- DefaultCredentialsProvider.builder().build();
- private static GlueSchemaRegistrySerializationFacade mockSerializationFacade;
-
- @BeforeAll
- static void setup() {
- metadata.put("test-key", "test-value");
- metadata.put(AWSSchemaRegistryConstants.TRANSPORT_METADATA_KEY, testTopic);
-
- configs.put(AWSSchemaRegistryConstants.AWS_REGION, "us-west-2");
- configs.put(AWSSchemaRegistryConstants.AWS_ENDPOINT, "https://test");
- configs.put(AWSSchemaRegistryConstants.SCHEMA_AUTO_REGISTRATION_SETTING, true);
- configs.put(AWSSchemaRegistryConstants.SCHEMA_NAME, schemaName);
- configs.put(AWSSchemaRegistryConstants.METADATA, metadata);
-
- userSchema =
- JsonDataWithSchema.builder(JSON_SCHEMA, "{\"f1\":\"iphone\",\"f2\":12}").build();
- userDefinedPojo =
- Car.builder()
- .make("Tesla")
- .model("3")
- .used(false)
- .miles(6000)
- .year(2021)
- .listedDate(new GregorianCalendar(2020, Calendar.FEBRUARY, 20).getTime())
- .purchaseDate(Date.from(Instant.parse("2020-01-01T00:00:00.000Z")))
- .owners(new String[] {"Harry", "Megan"})
- .serviceChecks(Arrays.asList(5000.0f, 10780.30f))
- .build();
-
- glueSchemaRegistryConfiguration = new GlueSchemaRegistryConfiguration(configs);
- mockSerializationFacade = new MockGlueSchemaRegistrySerializationFacade();
- }
-
- /** Test initialization works. */
- @Test
- void testForGeneric_withValidParams_succeeds() {
- assertThat(new GlueSchemaRegistryJsonSerializationSchema<>(testTopic, configs)).isNotNull();
- assertThat(new GlueSchemaRegistryJsonSerializationSchema<>(testTopic, configs))
- .isInstanceOf(GlueSchemaRegistryJsonSerializationSchema.class);
- }
-
- /**
- * Test whether serialize method for specific type JSON Schema data when compression is not
- * enabled works.
- */
- @Test
- void testSerializePOJO_withValidParams_withoutCompression_succeeds() {
- configs.put(AWSSchemaRegistryConstants.COMPRESSION_TYPE, NONE.name());
-
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder =
- new GlueSchemaRegistryJsonSchemaCoder(
- testTopic, configs, mockSerializationFacade, null);
-
- GlueSchemaRegistryJsonSerializationSchema<Car> glueSchemaRegistryJsonSerializationSchema =
- new GlueSchemaRegistryJsonSerializationSchema<>(glueSchemaRegistryJsonSchemaCoder);
-
- byte[] serializedData =
- glueSchemaRegistryJsonSerializationSchema.serialize(userDefinedPojo);
- assertThat(serializedData).isEqualTo(serializedBytes);
- }
-
- /**
- * Test whether serialize method for generic type JSON Schema data when compression is not
- * enabled works.
- */
- @Test
- void testSerializeGenericData_withValidParams_withoutCompression_succeeds() {
- configs.put(AWSSchemaRegistryConstants.COMPRESSION_TYPE, NONE.name());
-
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder =
- new GlueSchemaRegistryJsonSchemaCoder(
- testTopic, configs, mockSerializationFacade, null);
-
- GlueSchemaRegistryJsonSerializationSchema<JsonDataWithSchema>
- glueSchemaRegistryJsonSerializationSchema =
- new GlueSchemaRegistryJsonSerializationSchema<>(
- glueSchemaRegistryJsonSchemaCoder);
-
- byte[] serializedData = glueSchemaRegistryJsonSerializationSchema.serialize(userSchema);
- assertThat(serializedData).isEqualTo(serializedBytes);
- }
-
- /**
- * Test whether serialize method for specific type JSON Schema data when compression is enabled
- * works.
- */
- @Test
- void testSerializePOJO_withValidParams_withCompression_succeeds() {
- AWSSchemaRegistryConstants.COMPRESSION compressionType =
- AWSSchemaRegistryConstants.COMPRESSION.ZLIB;
- configs.put(AWSSchemaRegistryConstants.COMPRESSION_TYPE, compressionType.name());
-
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder =
- new GlueSchemaRegistryJsonSchemaCoder(
- testTopic, configs, mockSerializationFacade, null);
-
- GlueSchemaRegistryJsonSerializationSchema<Car> glueSchemaRegistryJsonSerializationSchema =
- new GlueSchemaRegistryJsonSerializationSchema<>(glueSchemaRegistryJsonSchemaCoder);
-
- byte[] serializedData =
- glueSchemaRegistryJsonSerializationSchema.serialize(userDefinedPojo);
- assertThat(serializedData).isEqualTo(serializedBytes);
- }
-
- /**
- * Test whether serialize method for generic type JSON Schema data when compression is enabled
- * works.
- */
- @Test
- void testSerializeGenericData_withValidParams_withCompression_succeeds() {
- AWSSchemaRegistryConstants.COMPRESSION compressionType =
- AWSSchemaRegistryConstants.COMPRESSION.ZLIB;
- configs.put(AWSSchemaRegistryConstants.COMPRESSION_TYPE, compressionType.name());
-
- GlueSchemaRegistryJsonSchemaCoder glueSchemaRegistryJsonSchemaCoder =
- new GlueSchemaRegistryJsonSchemaCoder(
- testTopic, configs, mockSerializationFacade, null);
-
- GlueSchemaRegistryJsonSerializationSchema<JsonDataWithSchema>
- glueSchemaRegistryJsonSerializationSchema =
- new GlueSchemaRegistryJsonSerializationSchema<>(
- glueSchemaRegistryJsonSchemaCoder);
-
- byte[] serializedData = glueSchemaRegistryJsonSerializationSchema.serialize(userSchema);
- assertThat(serializedData).isEqualTo(serializedBytes);
- }
-
- /** Test whether serialize method returns null when input object is null. */
- @Test
- void testSerialize_withNullObject_returnNull() {
- GlueSchemaRegistryJsonSerializationSchema<Car> glueSchemaRegistryJsonSerializationSchema =
- new GlueSchemaRegistryJsonSerializationSchema<>(testTopic, configs);
- assertThat(glueSchemaRegistryJsonSerializationSchema.serialize(null)).isNull();
- }
-
- private static class MockGlueSchemaRegistrySerializationFacade
- extends GlueSchemaRegistrySerializationFacade {
-
- public MockGlueSchemaRegistrySerializationFacade() {
- super(credentialsProvider, null, glueSchemaRegistryConfiguration, configs, null);
- }
-
- @Override
- public UUID getOrRegisterSchemaVersion(@NonNull AWSSerializerInput serializerInput) {
- return schemaVersionId;
- }
-
- @Override
- public byte[] serialize(
- DataFormat dataFormat, @NonNull Object data, @NonNull UUID schemaVersionId) {
- return serializedBytes;
- }
- }
-}
diff --git a/flink-formats/flink-json-glue-schema-registry/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/flink-formats/flink-json-glue-schema-registry/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
deleted file mode 100644
index 28999133c2b..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.
-
-org.apache.flink.util.TestLoggerExtension
\ No newline at end of file
diff --git a/flink-formats/flink-json-glue-schema-registry/src/test/resources/archunit.properties b/flink-formats/flink-json-glue-schema-registry/src/test/resources/archunit.properties
deleted file mode 100644
index 15be88c95ba..00000000000
--- a/flink-formats/flink-json-glue-schema-registry/src/test/resources/archunit.properties
+++ /dev/null
@@ -1,31 +0,0 @@
-#
-# 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.
-#
-
-# By default we allow removing existing violations, but fail when new violations are added.
-freeze.store.default.allowStoreUpdate=true
-
-# Enable this if a new (frozen) rule has been added in order to create the initial store and record the existing violations.
-#freeze.store.default.allowStoreCreation=true
-
-# Enable this to add allow new violations to be recorded.
-# NOTE: Adding new violations should be avoided when possible. If the rule was correct to flag a new
-# violation, please try to avoid creating the violation. If the violation was created due to a
-# shortcoming of the rule, file a JIRA issue so the rule can be improved.
-#freeze.refreeze=true
-
-freeze.store.default.path=archunit-violations
diff --git a/flink-formats/pom.xml b/flink-formats/pom.xml
index cca8901b771..c4ddf1acfa6 100644
--- a/flink-formats/pom.xml
+++ b/flink-formats/pom.xml
@@ -50,8 +50,6 @@ under the License.
<module>flink-orc-nohive</module>
<module>flink-hadoop-bulk</module>
<module>flink-format-common</module>
- <module>flink-avro-glue-schema-registry</module>
- <module>flink-json-glue-schema-registry</module>
</modules>
<!-- override these root dependencies as 'provided', so they don't end up
diff --git a/tools/azure-pipelines/build-apache-repo.yml b/tools/azure-pipelines/build-apache-repo.yml
index d28d4ee0264..3102d18acc7 100644
--- a/tools/azure-pipelines/build-apache-repo.yml
+++ b/tools/azure-pipelines/build-apache-repo.yml
@@ -54,8 +54,6 @@ variables:
SECRET_S3_BUCKET: $[variables.IT_CASE_S3_BUCKET]
SECRET_S3_ACCESS_KEY: $[variables.IT_CASE_S3_ACCESS_KEY]
SECRET_S3_SECRET_KEY: $[variables.IT_CASE_S3_SECRET_KEY]
- SECRET_GLUE_SCHEMA_ACCESS_KEY: $[variables.IT_CASE_GLUE_SCHEMA_ACCESS_KEY]
- SECRET_GLUE_SCHEMA_SECRET_KEY: $[variables.IT_CASE_GLUE_SCHEMA_SECRET_KEY]
stages:
# CI / PR triggered stage:
diff --git a/tools/azure-pipelines/e2e-template.yml b/tools/azure-pipelines/e2e-template.yml
index 53940667ba6..04041ca9a44 100644
--- a/tools/azure-pipelines/e2e-template.yml
+++ b/tools/azure-pipelines/e2e-template.yml
@@ -115,8 +115,6 @@ jobs:
IT_CASE_S3_BUCKET: $(SECRET_S3_BUCKET)
IT_CASE_S3_ACCESS_KEY: $(SECRET_S3_ACCESS_KEY)
IT_CASE_S3_SECRET_KEY: $(SECRET_S3_SECRET_KEY)
- IT_CASE_GLUE_SCHEMA_ACCESS_KEY: $(SECRET_GLUE_SCHEMA_ACCESS_KEY)
- IT_CASE_GLUE_SCHEMA_SECRET_KEY: $(SECRET_GLUE_SCHEMA_SECRET_KEY)
condition: and(succeeded(), not(eq(variables['SKIP'], '1')))
# upload debug artifacts
- task: PublishPipelineArtifact@1
diff --git a/tools/azure-pipelines/jobs-template.yml b/tools/azure-pipelines/jobs-template.yml
index 01629069fce..088977108f7 100644
--- a/tools/azure-pipelines/jobs-template.yml
+++ b/tools/azure-pipelines/jobs-template.yml
@@ -156,8 +156,6 @@ jobs:
IT_CASE_S3_BUCKET: $(SECRET_S3_BUCKET)
IT_CASE_S3_ACCESS_KEY: $(SECRET_S3_ACCESS_KEY)
IT_CASE_S3_SECRET_KEY: $(SECRET_S3_SECRET_KEY)
- IT_CASE_GLUE_SCHEMA_ACCESS_KEY: $(SECRET_GLUE_SCHEMA_ACCESS_KEY)
- IT_CASE_GLUE_SCHEMA_SECRET_KEY: $(SECRET_GLUE_SCHEMA_SECRET_KEY)
- task: PublishTestResults@2
condition: succeededOrFailed()
diff --git a/tools/ci/stage.sh b/tools/ci/stage.sh
index 4746fcef8d7..819868ef67a 100755
--- a/tools/ci/stage.sh
+++ b/tools/ci/stage.sh
@@ -102,8 +102,6 @@ flink-formats,\
flink-formats/flink-format-common,\
flink-formats/flink-avro-confluent-registry,\
flink-formats/flink-sql-avro-confluent-registry,\
-flink-formats/flink-avro-glue-schema-registry,\
-flink-formats/flink-json-glue-schema-registry,\
flink-formats/flink-avro,\
flink-formats/flink-sql-avro,\
flink-formats/flink-compress,\
diff --git a/tools/maven/suppressions.xml b/tools/maven/suppressions.xml
index b19d41e0ce5..303b4731f17 100644
--- a/tools/maven/suppressions.xml
+++ b/tools/maven/suppressions.xml
@@ -60,9 +60,6 @@ under the License.
<suppress
files="[\\/]target[\\/]"
checks=".*"/>
- <suppress
- files="org[\\/]apache[\\/]flink[\\/]formats[\\/]avro[\\/]glue[\\/]schema[\\/]registry[\\/]User.java"
- checks=".*"/>
<!-- suppress check for copied hive code -->
<suppress files="org[\\/]apache[\\/]hadoop[\\/]hive[\\/].*.java" checks=".*"/>
<suppress files="org[\\/]apache[\\/]hive[\\/]service[\\/]rpc[\\/]thrift[\\/].*.java" checks=".*"/>