[FLINK-24379] Add Glue Schema Registry Avro SQL format factory by fmorillo7694 · Pull Request #236 · apache/flink-connector-aws

fmorillo7694 · 2026-02-23T18:04:45Z

…glue)

Adds a Flink SQL / Table API format factory for AWS Glue Schema Registry with Avro support. Users can now use 'format' = 'avro-glue' in CREATE TABLE statements to read/write Avro data through GSR.

Based on the initial implementation by @nicusX. This PR extends that work with:

Shared config infrastructure (GlueFormatOptions, GlueFormatConfigBuilder)
Schema namespace bug fix (avro.namespace, avro.record-name options)
Schema fetch from registry (schema.fetchFromRegistry option)
ProjectableDecodingFormat support
forwardOptions() implementation
Rebase to 6.0-SNAPSHOT
JUnit 5 + AssertJ tests (ported from JUnit 4)
Property-based tests (jqwik)
Integration tests with mock GSR facades

Purpose of the change

For example: Implements the Table API for the Kinesis Source.

Verifying this change

Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment
Added unit tests
Manually verified by running the Kinesis connector on a local Flink cluster.

Significant changes

(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)

Dependencies have been added or upgraded
Public API has been changed (Public API is any class annotated with @Public(Evolving))
Serializers have been changed
New feature has been introduced
- If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)

End-to-end SQL test (real Glue Schema Registry)

GlueSchemaRegistryAvroSqlKinesisITCase replaces the former manual AvroGlueSqlE2E driver, following the existing GSR e2e convention (Localstack Kinesis data plane + real AWS Glue Schema Registry, @Tag("requires-aws-credentials"), skips cleanly without IT_CASE_GLUE_SCHEMA_ACCESS_KEY/SECRET_KEY). 7 scenarios: basic round-trip, custom namespace/record-name, schema.fetchFromRegistry, complex types (nested ROW/ARRAY/MAP/DECIMAL/TIMESTAMP/nullable), and BACKWARD / NONE / FULL compatibility enforcement. Run with the run-aws-end-to-end-tests profile.

Add the `protobuf-glue` Flink SQL format that serializes/deserializes RowData through AWS Glue Schema Registry, stacked on the Avro SQL format (PR apache#236). Mirrors the Avro module layout: format factory + options, RowType<->Protobuf converters, schema converter, RowData ser/de schemas, SQL uber-jar module, docs, and E2E test module. Applies the following review findings on top of the candidate implementation: - B4: encode/decode TIMESTAMP and TIMESTAMP_LTZ as epoch-millis int64, DATE/TIME as int32, and DECIMAL as its lossless BigDecimal text form, using the correct RowData accessors instead of getString() (which previously threw ClassCastException for these very common types). - C3: fail fast for genuinely unsupported complex types (ARRAY, MAP, MULTISET, ROW, RAW) instead of silently coercing them to `string`. - C4: sanitize column names to valid proto field identifiers and carry the original SQL name as the field's json_name, avoiding DescriptorValidationException at open() for names with spaces, hyphens, or a leading digit. - V2: pin protobuf-java and protobuf-java-util to a single <protobuf.version> property (3.25.x) to remove the version-skew risk between the two artifacts. Adds ProtobufTypeCoverageTest covering the temporal/decimal round-trips, field-name sanitization + json_name preservation, and the fail-fast path. Module builds green: 15 tests pass (mvn test, JDK 17), spotless clean. Known follow-ups (tracked in the GSR review, deferred): the SQL decode path still strips the fixed GSR header and rebuilds the reader descriptor from the local RowType rather than resolving the writer schema through the GSR deserialization facade (finding B1), so compression is not yet symmetric on read (C1) and proto3 implicit presence keeps null<->default round-trips lossy (C2).

Add the `json-glue` table format (SPI-discovered) that wraps Flink's stock JSON RowData ser/de with an AWS Glue Schema Registry envelope, plus the `flink-sql-json-glue-schema-registry` shaded uber-jar. Stacks on the Protobuf SQL format branch (which stacks on the Avro PR, apache#236), mirroring the established Avro/Protobuf module layout. Review fixes applied to JsonSchemaConverter (see reviews/glue-schema-registry-review.md): - JSON-escape field names (top-level and nested ROW) so a column containing a quote/backslash no longer yields malformed schema JSON. - Fail fast on genuinely unsupported types (MULTISET, RAW, INTERVAL, STRUCTURED, TIMESTAMP_WITH_TIME_ZONE, ...) instead of silently coercing to "string". - Emit "format" keywords for temporal types and "contentEncoding":"base64" for binary; constrain MAP keys to character types. - Emit a "required" array for NOT NULL fields; compose nullable unions directly instead of the brittle indexOf/substring re-parse. Tests: GlueSchemaRegistryJsonFormatFactoryTest (SPI + required-option validation), GsrJsonRoundTripPropertyTest, JsonRoundTripIntegrationTest, and a new JsonSchemaConverterTest (10 cases) covering escaping, fail-fast, formats, base64, map-key validation, required array, and nested rows. Module tests green under JDK 17; SQL uber-jar packages and merges the SPI Factory entry. Deferred (consistent with the Protobuf PR): SQL read path does not yet resolve the writer schema from GSR (no registry query / version-UUID resolution) and compression is not symmetric on read — left as the larger GSR-facade integration follow-up flagged in the review.