[FLINK-19667] Add AWS Glue Schema Registry integration #14737

mohitpali · 2021-01-24T02:34:25Z

What is the purpose of the change

The AWS Glue Schema Registry is a new feature of AWS Glue that allows you to centrally discover, control, and evolve data stream schemas. This request is to add a new format to launch an integration for Apache Flink with AWS Glue Schema Registry.

Brief change log

Added flink-avro-glue-schema-registry module under flink-formats
Added end-to-end test named flink-glue-schema-registry-test for the new module

Verifying this change

This change added tests and can be verified as follows:

Added integration tests for end-to-end deployment

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (yes)
The runtime per-record code paths (performance sensitive): (don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)

flinkbot · 2021-01-24T02:36:15Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 399f06e (Sun Jan 24 02:36:14 UTC 2021)

Warnings:

4 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2021-01-24T02:56:16Z

CI report:

2e05b89 UNKNOWN
288129b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

rmetzger · 2021-01-25T10:41:51Z

What's the relationship of this PR to #14490 ?

LinyuYao1021 · 2021-01-25T17:41:43Z

What's the relationship of this PR to #14490 ?

It's the same one but fixing the compiling error.

rmetzger · 2021-01-25T18:42:59Z

You don't need to open a PR for every fix, you can just keep (force) pushing to the branch. Can you close the old PR?

I'll review the PR tomorrow.

mohitpali · 2021-01-26T01:00:53Z

Apologies for the confusion, we have closed the other PR. We had to create another PR because two developers were working on it and hence the different login. I have included some CI compilation fixes in this PR and rebased.

rmetzger · 2021-01-27T14:41:10Z

Thanks a lot for the clarification

rmetzger

Thanks a lot for this big PR. I've made a first very rough pass over the code and commented on some issues. Depending on my level of confidence after a full review, I might ask another Flink committer to take a look as well.

What's your plan regarding documentation? Will this be done in a follow up PR, or in this one?

rmetzger · 2021-01-27T14:44:17Z

...ry-test/src/main/java/org.apache.flink.glue.schema.registry.test/GSRKinesisPubsubClient.java

@@ -0,0 +1,185 @@
+/*


It seems that the package name is not properly encoded into subdirectories.
Part of the directory name of this file is org.apache.flink.glue.schema.registry.test, but it should be org/apache/flink/glue/schema/registry/test. This might be difficult to see in some IDEs, as they are replacing this directory structure with the dot-notation.

rmetzger · 2021-01-27T14:50:48Z

.../src/main/java/org.apache.flink.glue.schema.registry.test/GlueSchemaRegistryExampleTest.java

+ // TODO: main thread needs to create job or CLI fails with:
+ // "The program didn't contain a Flink job. Perhaps you forgot to call execute() on the
+ // execution environment."
+ System.out.println("test finished");


flink-end-to-end-tests/test-scripts/test_glue_schema_registry.sh

rmetzger · 2021-01-27T16:17:04Z

...pache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializer.java

+
+ Schema schema;
+ try {
+ schema = (new Schema.Parser()).parse(schemaDefinition);


if I'm not mistaken, this parser initialization and schema parsing is done for every RegistryAvroDeserializationSchema.deserialize() call.
I guess this is necessary when deserializing GenericRecord Avro records, but for SpecificRecord we only need to deserialize the schema once?

The use of schema parsing is because GSR return its own defined Schema class from serialized byte array.

The question was focusing on the frequency of deserialisation. To improve performance can we deserialise schema for specific info once, and cache it? Or are we expecting the schema definition to change over time? What happens if the schema changes for a SpecificRecord? Is the idea that the Flink job would fail if the upstream data format changes in a non compatible way?

...rg/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderTest.java

flink-formats/flink-avro-glue-schema-registry/pom.xml

dannycranmer

Thank you for the contribution. I have taken a first pass at the PR.

dannycranmer · 2021-01-28T14:20:05Z

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml

+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-glue-schema-registry-test_${scala.binary.version}</artifactId>


Correct me if I am wrong, but I do not think you need additional artifacts per scala version. Suggest dropping _${scala.binary.version}

dannycranmer · 2021-01-28T14:23:24Z

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml

+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>${junit.version}</version>
+ <scope>compile</scope>


nit: you should not need <version> and <scope> tags here. I assume you meant to include junit as a compile scoped dependency (compile is default)

dannycranmer · 2021-01-28T14:25:10Z

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml

+ <aws.sdk.version>1.11.754</aws.sdk.version>
+ <aws.sdkv2.version>2.15.32</aws.sdkv2.version>


nit: Consider updating, at the time of writing:

v1 @ 1.11.943

v2 @ 2.15.70

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml

dannycranmer · 2021-01-28T14:29:22Z

...ry-test/src/main/java/org.apache.flink.glue.schema.registry.test/GSRKinesisPubsubClient.java

+ this.properties = properties;
+ }
+
+ public void createTopic(String stream, int shards, Properties props) throws Exception {


createStream? I would consider updating the method name or add Javadoc to indicate that an existing stream will be deleted

...pache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSerializationSchema.java

flink-formats/flink-avro-glue-schema-registry/pom.xml

dannycranmer · 2021-01-28T15:47:30Z

flink-formats/flink-avro-glue-schema-registry/pom.xml

+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>


This should probably be <scope>provided</scope>

It'll cause ClassNotFoundException if adding this scope.

Where are you seeing ClassNotFoundExcception? At runtime the Class would be provided by the Flink cluster. You may see an issue running in standalone mode. The problem with not making this provided is that when building an uber jar for an app, it could bundle additional unnecessary Flink code into the jar. This would bloat the jar size and classpath.

What is fixed? I do not see any changes?

The change may be lost during rebasing. Will add again in next commit.

dannycranmer · 2021-01-28T15:51:37Z

flink-formats/flink-avro-glue-schema-registry/pom.xml

+ <enforcer.skip>true</enforcer.skip>
+ </properties>
+
+ <dependencies>


I have taken a look at the dependency footprint of this module and it looks like there is too much pulled in:

Why do we need Kafka dependencies?

+- org.apache.kafka:connect-json:jar:2.5.0:compile

+- org.apache.kafka:connect-api:jar:2.5.0:compile

+- org.apache.kafka:kafka-streams:jar:2.5.0:compile

+- org.apache.kafka:kafka-clients:jar:2.5.0:compile

Pulling lombok as a compile dependency looks wrong, is this scoped correctly in upstream module?

+- org.projectlombok:lombok:jar:1.18.2:compile

\- org.projectlombok:lombok-utils:jar:1.18.12:compile

As mentioned, we should use the standard junit framework for flink:

+- org.junit.jupiter:junit-jupiter-api:jar:5.6.2:test

[INFO] --- maven-dependency-plugin:3.1.1:tree (default-cli) @ flink-avro-glue-schema-registry --- [INFO] org.apache.flink:flink-avro-glue-schema-registry:jar:1.13-SNAPSHOT [INFO] +- org.apache.flink:flink-core:jar:1.13-SNAPSHOT:provided [INFO] | +- org.apache.flink:flink-annotations:jar:1.13-SNAPSHOT:provided [INFO] | +- org.apache.flink:flink-metrics-core:jar:1.13-SNAPSHOT:provided [INFO] | +- org.apache.flink:flink-shaded-asm-7:jar:7.1-12.0:provided [INFO] | +- org.apache.commons:commons-lang3:jar:3.3.2:compile [INFO] | +- com.esotericsoftware.kryo:kryo:jar:2.24.0:provided [INFO] | | \- com.esotericsoftware.minlog:minlog:jar:1.2:provided [INFO] | +- commons-collections:commons-collections:jar:3.2.2:provided [INFO] | +- org.apache.commons:commons-compress:jar:1.20:compile [INFO] | \- org.apache.flink:flink-shaded-guava:jar:18.0-12.0:compile [INFO] +- org.apache.flink:flink-avro:jar:1.13-SNAPSHOT:compile [INFO] | \- org.apache.avro:avro:jar:1.10.0:compile [INFO] | +- com.fasterxml.jackson.core:jackson-core:jar:2.12.1:compile [INFO] | \- com.fasterxml.jackson.core:jackson-databind:jar:2.12.1:compile [INFO] | \- com.fasterxml.jackson.core:jackson-annotations:jar:2.12.1:compile [INFO] +- org.apache.flink:flink-streaming-java_2.11:jar:1.13-SNAPSHOT:compile [INFO] | +- org.apache.flink:flink-file-sink-common:jar:1.13-SNAPSHOT:compile [INFO] | +- org.apache.flink:flink-runtime_2.11:jar:1.13-SNAPSHOT:compile [INFO] | | +- org.apache.flink:flink-queryable-state-client-java:jar:1.13-SNAPSHOT:compile [INFO] | | +- org.apache.flink:flink-hadoop-fs:jar:1.13-SNAPSHOT:compile [INFO] | | +- commons-io:commons-io:jar:2.7:compile [INFO] | | +- org.apache.flink:flink-shaded-netty:jar:4.1.49.Final-12.0:compile [INFO] | | +- org.apache.flink:flink-shaded-jackson:jar:2.10.1-12.0:compile [INFO] | | +- org.apache.flink:flink-shaded-zookeeper-3:jar:3.4.14-12.0:compile [INFO] | | +- org.javassist:javassist:jar:3.24.0-GA:compile [INFO] | | +- org.scala-lang:scala-library:jar:2.11.12:compile [INFO] | | +- com.typesafe.akka:akka-actor_2.11:jar:2.5.21:compile [INFO] | | | +- com.typesafe:config:jar:1.3.0:compile [INFO] | | | \- org.scala-lang.modules:scala-java8-compat_2.11:jar:0.7.0:compile [INFO] | | +- com.typesafe.akka:akka-stream_2.11:jar:2.5.21:compile [INFO] | | | +- org.reactivestreams:reactive-streams:jar:1.0.2:compile [INFO] | | | \- com.typesafe:ssl-config-core_2.11:jar:0.3.7:compile [INFO] | | | \- org.scala-lang.modules:scala-parser-combinators_2.11:jar:1.1.1:compile [INFO] | | +- com.typesafe.akka:akka-protobuf_2.11:jar:2.5.21:compile [INFO] | | +- com.typesafe.akka:akka-slf4j_2.11:jar:2.5.21:compile [INFO] | | +- org.clapper:grizzled-slf4j_2.11:jar:1.3.2:compile [INFO] | | +- com.github.scopt:scopt_2.11:jar:3.5.0:compile [INFO] | | +- org.xerial.snappy:snappy-java:jar:1.1.4:compile [INFO] | | +- com.twitter:chill_2.11:jar:0.7.6:compile [INFO] | | | \- com.twitter:chill-java:jar:0.7.6:compile [INFO] | | \- org.lz4:lz4-java:jar:1.6.0:compile [INFO] | +- org.apache.flink:flink-java:jar:1.13-SNAPSHOT:compile [INFO] | \- org.apache.commons:commons-math3:jar:3.5:compile [INFO] +- org.apache.flink:flink-clients_2.11:jar:1.13-SNAPSHOT:compile [INFO] | +- org.apache.flink:flink-optimizer_2.11:jar:1.13-SNAPSHOT:compile [INFO] | \- commons-cli:commons-cli:jar:1.3.1:compile [INFO] +- software.amazon.glue:schema-registry-serde:jar:1.0.0:compile [INFO] | +- software.amazon.glue:schema-registry-common:jar:1.0.0:compile [INFO] | | +- software.amazon.awssdk:glue:jar:2.15.32:compile [INFO] | | | +- software.amazon.awssdk:protocol-core:jar:2.15.32:compile [INFO] | | | +- software.amazon.awssdk:auth:jar:2.15.32:compile [INFO] | | | | \- software.amazon.eventstream:eventstream:jar:1.0.1:compile [INFO] | | | +- software.amazon.awssdk:http-client-spi:jar:2.15.32:compile [INFO] | | | +- software.amazon.awssdk:regions:jar:2.15.32:compile [INFO] | | | +- software.amazon.awssdk:aws-core:jar:2.15.32:compile [INFO] | | | +- software.amazon.awssdk:metrics-spi:jar:2.15.32:compile [INFO] | | | +- software.amazon.awssdk:apache-client:jar:2.15.32:runtime [INFO] | | | | +- org.apache.httpcomponents:httpclient:jar:4.5.3:runtime [INFO] | | | | | +- commons-logging:commons-logging:jar:1.1.3:runtime [INFO] | | | | | \- commons-codec:commons-codec:jar:1.13:runtime [INFO] | | | | \- org.apache.httpcomponents:httpcore:jar:4.4.6:runtime [INFO] | | | \- software.amazon.awssdk:netty-nio-client:jar:2.15.32:runtime [INFO] | | | +- io.netty:netty-codec-http:jar:4.1.53.Final:runtime [INFO] | | | +- io.netty:netty-codec-http2:jar:4.1.53.Final:runtime [INFO] | | | +- io.netty:netty-codec:jar:4.1.53.Final:runtime [INFO] | | | +- io.netty:netty-transport:jar:4.1.53.Final:runtime [INFO] | | | | \- io.netty:netty-resolver:jar:4.1.53.Final:runtime [INFO] | | | +- io.netty:netty-common:jar:4.1.53.Final:runtime [INFO] | | | +- io.netty:netty-buffer:jar:4.1.53.Final:runtime [INFO] | | | +- io.netty:netty-handler:jar:4.1.53.Final:runtime [INFO] | | | +- io.netty:netty-transport-native-epoll:jar:linux-x86_64:4.1.53.Final:runtime [INFO] | | | | \- io.netty:netty-transport-native-unix-common:jar:4.1.53.Final:runtime [INFO] | | | \- com.typesafe.netty:netty-reactive-streams-http:jar:2.0.4:runtime [INFO] | | | \- com.typesafe.netty:netty-reactive-streams:jar:2.0.4:runtime [INFO] | | +- software.amazon.awssdk:aws-json-protocol:jar:2.15.30:compile [INFO] | | +- software.amazon.awssdk:cloudwatch:jar:2.15.30:compile [INFO] | | | \- software.amazon.awssdk:aws-query-protocol:jar:2.15.30:compile [INFO] | | +- software.amazon.awssdk:sdk-core:jar:2.15.30:compile [INFO] | | | \- software.amazon.awssdk:profiles:jar:2.15.30:compile [INFO] | | +- org.apache.kafka:kafka-clients:jar:2.5.0:compile [INFO] | | | \- com.github.luben:zstd-jni:jar:1.4.4-7:compile [INFO] | | +- org.apache.kafka:kafka-streams:jar:2.5.0:compile [INFO] | | | +- org.apache.kafka:connect-json:jar:2.5.0:compile [INFO] | | | | +- org.apache.kafka:connect-api:jar:2.5.0:compile [INFO] | | | | \- com.fasterxml.jackson.datatype:jackson-datatype-jdk8:jar:2.12.1:compile [INFO] | | | \- org.rocksdb:rocksdbjni:jar:5.18.3:compile [INFO] | | \- com.google.guava:guava:jar:29.0-jre:compile [INFO] | | +- com.google.guava:failureaccess:jar:1.0.1:compile [INFO] | | +- com.google.guava:listenablefuture:jar:9999.0-empty-to-avoid-conflict-with-guava:compile [INFO] | | +- org.checkerframework:checker-qual:jar:2.11.1:compile [INFO] | | +- com.google.errorprone:error_prone_annotations:jar:2.3.4:compile [INFO] | | \- com.google.j2objc:j2objc-annotations:jar:1.3:compile [INFO] | +- software.amazon.awssdk:arns:jar:2.15.26:compile [INFO] | | +- software.amazon.awssdk:annotations:jar:2.15.26:compile [INFO] | | \- software.amazon.awssdk:utils:jar:2.15.26:compile [INFO] | +- org.projectlombok:lombok:jar:1.18.2:compile [INFO] | \- org.projectlombok:lombok-utils:jar:1.18.12:compile [INFO] +- org.junit.jupiter:junit-jupiter-api:jar:5.6.2:test [INFO] | +- org.apiguardian:apiguardian-api:jar:1.1.0:test [INFO] | +- org.opentest4j:opentest4j:jar:1.2.0:test [INFO] | \- org.junit.platform:junit-platform-commons:jar:1.6.2:test [INFO] +- org.junit.jupiter:junit-jupiter-params:jar:5.6.2:test [INFO] +- org.mockito:mockito-junit-jupiter:jar:2.21.0:test [INFO] +- org.slf4j:slf4j-api:jar:1.7.15:provided [INFO] +- org.apache.flink:flink-test-utils-junit:jar:1.13-SNAPSHOT:test [INFO] +- org.apache.flink:force-shading:jar:1.13-SNAPSHOT:compile [INFO] +- com.google.code.findbugs:jsr305:jar:1.3.9:compile [INFO] +- junit:junit:jar:4.12:test [INFO] | \- org.hamcrest:hamcrest-core:jar:1.3:test [INFO] +- org.mockito:mockito-core:jar:2.21.0:test [INFO] | +- net.bytebuddy:byte-buddy:jar:1.8.15:test [INFO] | +- net.bytebuddy:byte-buddy-agent:jar:1.8.15:test [INFO] | \- org.objenesis:objenesis:jar:2.1:provided [INFO] +- org.powermock:powermock-module-junit4:jar:2.0.4:test [INFO] | \- org.powermock:powermock-module-junit4-common:jar:2.0.4:test [INFO] | +- org.powermock:powermock-reflect:jar:2.0.4:test [INFO] | \- org.powermock:powermock-core:jar:2.0.4:test [INFO] +- org.powermock:powermock-api-mockito2:jar:2.0.4:test [INFO] | \- org.powermock:powermock-api-support:jar:2.0.4:test [INFO] +- org.hamcrest:hamcrest-all:jar:1.3:test [INFO] +- org.apache.logging.log4j:log4j-slf4j-impl:jar:2.12.1:test [INFO] +- org.apache.logging.log4j:log4j-api:jar:2.12.1:test [INFO] +- org.apache.logging.log4j:log4j-core:jar:2.12.1:test [INFO] \- org.apache.logging.log4j:log4j-1.2-api:jar:2.12.1:test

The enforcer check skip property has been removed. Currently, version of all dependencies with convergence error have been specifically defined in Flink-GSR module and its e2e test module. Once the new version of GSR package with reorganized dependencies is released to maven, the version definition can be removed.

This comment looks misplaced. This is not related to the enforcer skip, it is looking at the transitive dependency chain. Did you reply to the wrong thread?

The dependency chain is fixed in GSR package but it'll need some time to release. Once it's out, it should also fix the enforcer check issue.

ok, so we have removed the dependency on Kafka in the new version? What is the ECD for the new version?

We will fix this in a follow up

dannycranmer · 2021-02-17T14:40:57Z

I am seeing a test failure when running mvn clean install on the flink-avro-glue-schema-registry:

[ERROR] Errors:
[ERROR]   GlueSchemaRegistryAvroSerializationSchemaTest.<init>:58 » IllegalArgument glue...
[ERROR]   GlueSchemaRegistryAvroSerializationSchemaTest.<init>:58 » IllegalArgument glue...
[ERROR]   GlueSchemaRegistryAvroSerializationSchemaTest.<init>:58 » IllegalArgument glue...
[ERROR]   GlueSchemaRegistryAvroSerializationSchemaTest.<init>:58 » IllegalArgument glue...
[ERROR]   GlueSchemaRegistryAvroSerializationSchemaTest.<init>:58 » IllegalArgument glue...
[INFO]
[ERROR] Tests run: 19, Failures: 0, Errors: 5, Skipped: 0

Looks like glueSchemaRegistryConfiguration is null

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml

...ry-test/src/main/java/org/apache/flink/glue/schema/registry/test/GSRKinesisPubsubClient.java

...test/src/main/java/org/apache/flink/glue/schema/registry/test/GlueSchemaRegistryExample.java

dannycranmer · 2021-02-17T14:26:30Z

flink-formats/flink-avro-glue-schema-registry/pom.xml

+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>


What is fixed? I do not see any changes?

dannycranmer · 2021-02-17T14:28:07Z

flink-formats/flink-avro-glue-schema-registry/pom.xml

+ <enforcer.skip>true</enforcer.skip>
+ </properties>
+
+ <dependencies>


This comment looks misplaced. This is not related to the enforcer skip, it is looking at the transitive dependency chain. Did you reply to the wrong thread?

dannycranmer · 2021-02-17T14:28:44Z

...va/org/apache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoder.java

+
+ @Override
+ public void writeSchema(Schema schema, OutputStream out) throws IOException {
+ byte[] data = ((ByteArrayOutputStream) out).toByteArray();


ok, please add the Preconditions check

dannycranmer · 2021-02-17T14:35:15Z

...pache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderProvider.java

+
+import org.apache.flink.formats.avro.SchemaCoder;
+
+import lombok.NonNull;


Yes I meant you are using lombok rather than javax.annotation. But I think this is not needed since Flink coding standards say everything is nonnull by default:

https://flink.apache.org/contributing/code-style-and-quality-common.html#nullability-of-the-mutable-parts

This also opens the question as to why the transportName is not @NonNull? Is it @Nullable?

Please remove the annotations unless there is a good reason to keep them.

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml

flink-end-to-end-tests/test-scripts/test_glue_schema_registry.sh

zentol · 2021-02-17T19:30:51Z

...e-schema-registry/src/test/java/org/apache/flink/formats/avro/glue/schema/registry/User.java

+
+@SuppressWarnings("all")
+@org.apache.avro.specific.AvroGenerated
+public class User extends org.apache.avro.specific.SpecificRecordBase


Do we actually need this generated file to be in the source, or could we just generate it on the fly like we do for flink-avro?

Currently, there's no avro file can be used directly. So this file is still needed

But isn't this file generated from the schema defined in flink-formats/flink-avro-glue-schema-registry/src/test/java/resources/avro/user.avsc?
The avro-maven-plugin can generated this User.java file based on user.avsc

I'm okay with addressing this in a follow up PR if you prefer.

Okay I misunderstood what the dependency does. Will address this in follow up PR.

zentol · 2021-02-17T19:31:59Z

...he/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryOutputStreamSerializerTest.java

+import static org.hamcrest.Matchers.instanceOf;
+
+/** Tests for {@link GlueSchemaRegistryOutputStreamSerializer}. */
+public class GlueSchemaRegistryOutputStreamSerializerTest {


It is common for all test class to extend TestLogger

...e/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializerTest.java

...pache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderProvider.java

dannycranmer · 2021-02-18T22:09:55Z

As a follow up we should add support for SQL client and Table API by:

Adding a shaded module for SQL client, similar to flink-sql-avro-confluent-registry
Create a FormatFactory
Add a service

flink-end-to-end-tests/test-scripts/test_glue_schema_registry.sh

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml

LinyuYao1021 · 2021-03-09T09:36:52Z

SECRET_GLUE_SCHEMA_ACCESS_KEY: $[variables.IT_CASE_GLUE_SCHEMA_ACCESS_KEY]
SECRET_GLUE_SCHEMA_SECRET_KEY: $[variables.IT_CASE_GLUE_SCHEMA_SECRET_KEY]

Added

LinyuYao1021 · 2021-03-09T17:04:15Z

Hi Robert, main CI passed now. Would you please bring the latest commit to your personal CI to verify it and then we can close this PR? @rmetzger

rmetzger · 2021-03-09T17:55:07Z

Have you setup your CI with the password as well, and verified the change?

LinyuYao1021 · 2021-03-09T18:00:16Z

Have you setup your CI with the password as well, and verified the change?

Not yet, I haven't used Azure before. Could you quickly walk me through what to do to set up CI?

dannycranmer · 2021-03-09T18:01:26Z

I also need to set this up. I was planning on taking a look tomorrow morning. Let me try to setup my personal Azure and verify your change. I will update you tomorrow.

rmetzger · 2021-03-09T18:38:35Z

Here's how to setup azure: https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository

rmetzger · 2021-03-09T18:39:31Z

https://dev.azure.com/rmetzger/Flink/_build/results?buildId=8955&view=results

rmetzger · 2021-03-09T21:27:35Z

It failed again: https://dev.azure.com/rmetzger/Flink/_build/results?buildId=8955&view=logs&j=9401bf33-03c4-5a24-83fe-e51d75db73ef&t=72901ab2-7cd0-57be-82b1-bca51de20fba

LinyuYao1021 · 2021-03-09T21:54:43Z

It failed again: https://dev.azure.com/rmetzger/Flink/_build/results?buildId=8955&view=logs&j=9401bf33-03c4-5a24-83fe-e51d75db73ef&t=72901ab2-7cd0-57be-82b1-bca51de20fba

It's still because credentials can't be extracted. How did you success last time?

dannycranmer · 2021-03-10T08:37:02Z

I have setup my pipeline, running master to verify it works, then I will run your GSR branch:

https://dev.azure.com/dannycranmer/Flink/_build/results?buildId=4&view=results

LinyuYao1021 · 2021-03-10T09:59:09Z

This is my pipeline, directly to run my commit to see what's the problem:

-https://dev.azure.com/yaolinyu3547/PrivateFlink/_build/results?buildId=5&view=results

dannycranmer · 2021-03-11T10:42:59Z

Current status is that my new Azure account is blocked waiting for limit increase to run parallel builds. Until this is complete, I cannot verify the e2e tests. I have sent an email to azure as described in the docs, and am waiting for a response.

rmetzger · 2021-03-11T11:21:33Z

I was concerned that this would happen to you. A new hire in our company is facing the same issue.
Looks like Azure is having some issues with cryptocurrency mining on their Infra, but they lack the proper infra to detect it.

dannycranmer · 2021-03-11T13:49:33Z

Thanks @rmetzger. GSR branch running:

https://dev.azure.com/georgeryan1322/Flink/_build/results?buildId=351&view=results

dannycranmer · 2021-03-11T17:39:52Z

Failed, tweaking the test and retry:

https://dev.azure.com/georgeryan1322/Flink/_build/results?buildId=352&view=results

dannycranmer · 2021-03-15T14:13:59Z

flink-end-to-end-tests/test-scripts/common.sh

@@ -364,6 +364,7 @@ function check_logs_for_errors {
 | grep -v "HeapDumpOnOutOfMemoryError" \
 | grep -v "error_prone_annotations" \
 | grep -v "Error sending fetch request" \
+ | grep -v "WARN akka.remote.ReliableDeliverySupervisor" \


@LinyuYao1021 Do we still need this change?

This is to avoid failure under this scenario:

2021-03-11T23:25:41.9106886Z Mar 11 23:25:41 2021-03-11 23:25:39,736 WARN akka.remote.ReliableDeliverySupervisor [] - Association with remote system [akka.tcp:https://[email protected]:37981] has failed, address is now gated for [50] ms. Reason: [Disassociated] 2021-03-11T23:25:41.9108202Z Mar 11 23:25:41 2021-03-11 23:25:39,747 WARN akka.remote.ReliableDeliverySupervisor [] - Association with remote system [akka.tcp:https://[email protected]:37839] has failed, address is now gated for [50] ms. Reason: [Disassociated] 2021-03-11T23:25:41.9109453Z Mar 11 23:25:41 2021-03-11 23:25:40,010 WARN akka.remote.transport.netty.NettyTransport [] - Remote connection to [null] failed with java.net.ConnectException: Connection refused: /10.1.0.4:37839 2021-03-11T23:25:41.9111511Z Mar 11 23:25:41 2021-03-11 23:25:40,010 WARN akka.remote.ReliableDeliverySupervisor [] - Association with remote system [akka.tcp:https://[email protected]:***@10.1.0.4:37839]] Caused by: [java.net.ConnectException: Connection refused: /10.1.0.4:37839]

We already ignore grep -v "WARN akka.remote.transport.netty.NettyTransport"

@LinyuYao1021 Do we still need this change?

Yes, it skips the akka exception check for logs.

dannycranmer · 2021-03-15T16:38:24Z

CI is passing with and without WS credentials. I will merge this now:

https://dev.azure.com/georgeryan1322/Flink/_build/results?buildId=360&view=results

rmetzger · 2021-03-17T14:23:51Z

Very nice! Thanks a lot for your efforts!

jiamo · 2021-03-22T08:52:10Z

A little question. We know with this repe https://github.com/awslabs/aws-glue-data-catalog-client-for-apache-hive-metastore .AWS EMR Hive can seamless talk with glue meta.
But when use flink hive. It use the original metastore client.
Is It possible to make an option that flink-connector-hive can talk with glue meta.

dannycranmer · 2021-03-22T16:39:06Z

@jiamo this is currently not possible out of the box, but can be achieved with some tweaks. Once you have Flink instantiating and using the Glue Data Catalog client, you would still need to implement property translation to sources/sinks etc (Kinesis/Kafka etc).

What is your use-case?

jiamo · 2021-03-23T03:40:57Z

My use-case: use flink-sql to read data from EMR managed hive (while the table content was s3 files and partition by day)

MartijnVisser · 2022-10-06T20:24:21Z

@mohitpali Do you think you could help with updating the dependencies of the Glue Schema Registry integrations in Flink? These are already quite outdated and it would be nice if we could update them to the latest supported versions.

dannycranmer · 2022-10-06T21:10:16Z

@MartijnVisser I will reach out to the team, if they cannot help I will find someone.

MartijnVisser · 2022-10-06T21:46:03Z

@dannycranmer Thanks!

dannycranmer · 2022-10-11T07:41:51Z

@MartijnVisser this will be picked up by hlteoh37 (for some reason I cannot mention him)

hlteoh37 · 2022-10-11T20:28:51Z

@MartijnVisser @dannycranmer Have picked it up here: https://issues.apache.org/jira/browse/FLINK-29574

rmetzger added the review=description? label Jan 24, 2021

rmetzger added the component=Formats label Jan 24, 2021

mohitpali force-pushed the master branch from 399f06e to bd61789 Compare January 24, 2021 07:28

mohitpali force-pushed the master branch from bd61789 to d6f06f0 Compare January 26, 2021 00:54

rmetzger requested changes Jan 27, 2021

View reviewed changes

dannycranmer self-requested a review January 28, 2021 14:08

dannycranmer requested changes Jan 28, 2021

View reviewed changes

LinyuYao1021 force-pushed the master branch 4 times, most recently from 2e05b89 to 81adc9c Compare February 17, 2021 11:40

dannycranmer requested changes Feb 17, 2021

View reviewed changes

zentol reviewed Feb 17, 2021

View reviewed changes

...e/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryInputStreamDeserializerTest.java Show resolved Hide resolved

LinyuYao1021 force-pushed the master branch from 81adc9c to 56566c3 Compare February 18, 2021 00:57

dannycranmer reviewed Feb 18, 2021

View reviewed changes

...pache/flink/formats/avro/glue/schema/registry/GlueSchemaRegistryAvroSchemaCoderProvider.java Outdated Show resolved Hide resolved

LinyuYao1021 force-pushed the master branch from 56566c3 to 061978a Compare February 25, 2021 17:27

dannycranmer reviewed Feb 26, 2021

View reviewed changes

flink-end-to-end-tests/test-scripts/test_glue_schema_registry.sh Show resolved Hide resolved

dannycranmer reviewed Feb 26, 2021

View reviewed changes

flink-end-to-end-tests/flink-glue-schema-registry-test/pom.xml Show resolved Hide resolved

LinyuYao1021 force-pushed the master branch from 061978a to 46aa624 Compare March 1, 2021 12:52

LinyuYao1021 force-pushed the master branch from bafd3a9 to c4eb439 Compare March 11, 2021 21:02

[FLINK-19667] Add AWS Glue Schema Registry integration

288129b

LinyuYao1021 force-pushed the master branch from c4eb439 to 288129b Compare March 12, 2021 02:14

dannycranmer reviewed Mar 15, 2021

View reviewed changes

dannycranmer merged commit b77afd0 into apache:master Mar 15, 2021

		<aws.sdk.version>1.11.754</aws.sdk.version>
		<aws.sdkv2.version>2.15.32</aws.sdkv2.version>


		import org.apache.flink.formats.avro.SchemaCoder;

		import lombok.NonNull;

[FLINK-19667] Add AWS Glue Schema Registry integration #14737

[FLINK-19667] Add AWS Glue Schema Registry integration #14737

Conversation

mohitpali commented Jan 24, 2021

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jan 24, 2021

Automated Checks

Review Progress

flinkbot commented Jan 24, 2021 • edited Loading

CI report:

rmetzger commented Jan 25, 2021

LinyuYao1021 commented Jan 25, 2021

rmetzger commented Jan 25, 2021

mohitpali commented Jan 26, 2021 • edited Loading

rmetzger commented Jan 27, 2021

rmetzger left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dannycranmer left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LinyuYao1021 Feb 17, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dannycranmer commented Feb 17, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dannycranmer commented Feb 18, 2021

LinyuYao1021 commented Mar 9, 2021

LinyuYao1021 commented Mar 9, 2021 • edited Loading

rmetzger commented Mar 9, 2021

LinyuYao1021 commented Mar 9, 2021

dannycranmer commented Mar 9, 2021

rmetzger commented Mar 9, 2021

rmetzger commented Mar 9, 2021

rmetzger commented Mar 9, 2021

LinyuYao1021 commented Mar 9, 2021

dannycranmer commented Mar 10, 2021

LinyuYao1021 commented Mar 10, 2021 • edited Loading

dannycranmer commented Mar 11, 2021

rmetzger commented Mar 11, 2021

dannycranmer commented Mar 11, 2021

dannycranmer commented Mar 11, 2021

Choose a reason for hiding this comment

flinkbot commented Jan 24, 2021 •

edited

Loading

mohitpali commented Jan 26, 2021 •

edited

Loading

LinyuYao1021 Feb 17, 2021 •

edited

Loading

dannycranmer commented Feb 17, 2021 •

edited

Loading

LinyuYao1021 commented Mar 9, 2021 •

edited

Loading

LinyuYao1021 commented Mar 10, 2021 •

edited

Loading

dannycranmer commented Mar 15, 2021 •

edited

Loading

jiamo commented Mar 22, 2021 •

edited

Loading