[FLINK-29549]- Flink Glue Catalog integration

[fmorillo7694]

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)

[Samrat002]

Thanks for the contribution. A few points to note:

It looks like most of the code has already been reviewed across multiple PRs.

Concerns:

1. The current implementation exposes low-level Glue details (e.g., support for lowercase table names) directly through the Flink Catalog. Is this acceptable? I recommend raising this topic with the community. In my opinion, it would be better to encapsulate such Glue-specific behavior and avoid exposing it directly via the Flink Catalog interface. The catalog's behavior should remain consistent with other catalogs, with differences controlled via configuration only.

2. Please consider implementing the basic configuration options defined in FLIP-277. If that’s not feasible in this PR, a fast follow-up would be valuable, especially for users relying on different credential modes.

3. It seems that some .idea folder files have been committed. Please remove them from the PR.

Cheers,
Samrat

[fmorillo7694]

Thanks for the contribution. A few points to note:

It looks like most of the code has already been reviewed across multiple PRs.

Concerns:

1. The current implementation exposes low-level Glue details (e.g., support for lowercase table names) directly through the Flink Catalog. Is this acceptable? I recommend raising this topic with the community. In my opinion, it would be better to encapsulate such Glue-specific behavior and avoid exposing it directly via the Flink Catalog interface. The catalog's behavior should remain consistent with other catalogs, with differences controlled via configuration only.

2. Please consider implementing the basic configuration options defined in [FLIP-277](https://cwiki.apache.org/confluence/display/FLINK/FLIP-277). If that’s not feasible in this PR, a fast follow-up would be valuable, especially for users relying on different credential modes.

3. It seems that some `.idea` folder files have been committed. Please remove them from the PR.

Cheers, Samrat

Hey Samrat. How could we encapsulate this specific issues? From user perspective we are already limiting them creating tables and databases with uppercase. In regards of the schema, we encapsulate storing the actual original columnNames in column parameters in Glue, so even though user sees their schema in glue with lower case (default for glue). we actually leverage the original column name.

2. Yes right after we get the first release delivered we will add the remainder authentication, and partitions support and alter support

3. Fixed!

[Samrat002]

Hey Samrat. How could we encapsulate this specific issues? From user perspective we are already limiting them creating tables and databases with uppercase. In regards of the schema, we encapsulate storing the actual original columnNames in column parameters in Glue, so even though user sees their schema in glue with lower case (default for glue). we actually leverage the original column name.

Thanks for the detailed response.

Can you help me understand the technical complexity to support CaseSensitivity from GlueCatalog?

This is a deviation.
I believe understanding how other connectors or catalogs handle mismatches, there can be a couple of ways to handle it

1. Surfacing the mismatch to the end user with constraints
2. Handling it in FlinkCatalog to provide consistent behaviour.

We can start a thread in the community to discuss this approach.

I am fine with either of the approaches as long as the community agrees on it.

[nicusX]

@Samrat002 @FranMorilloAWS about case sensitivity, these are the logical rules I suggest:

1. the actual case should preserved for all supported objects and used consistently in Flink interface, e.g. inCREATE TABLE... but also SHOW TABLE, SHOW COLUMNS ... etc. From a user's perspective, the way object names are actually stored in Glue should not be a concern.
2. The additional metadata should be stored in Glue in a consistent way for all object types. Parameters seems to be available for Database, Tables, and Columns in Glue. A problem arises for Functions which don't have any Parameters if I am not mistaken.
3. The additional metadata should not leak into Flink user interface. The "originalName" should not pop up in a SHOW TABLES or SHOW COLUMNS
4. The fact Glue UI is limited and only shows lowercase names is not a concern of this component's

I would add that the way the additional metadata (the original name and any additional info required) is stored in Glue should be clearly documented, in case a user want's to build their own external interface to extract or update information in the Glue Catalog.

[fmorillo7694]

Thanks for the contribution. A few points to note:

It looks like most of the code has already been reviewed across multiple PRs.

Concerns:

1. The current implementation exposes low-level Glue details (e.g., support for lowercase table names) directly through the Flink Catalog. Is this acceptable? I recommend raising this topic with the community. In my opinion, it would be better to encapsulate such Glue-specific behavior and avoid exposing it directly via the Flink Catalog interface. The catalog's behavior should remain consistent with other catalogs, with differences controlled via configuration only.

2. Please consider implementing the basic configuration options defined in [FLIP-277](https://cwiki.apache.org/confluence/display/FLINK/FLIP-277). If that’s not feasible in this PR, a fast follow-up would be valuable, especially for users relying on different credential modes.

3. It seems that some `.idea` folder files have been committed. Please remove them from the PR.

Cheers, Samrat

Hello @Samrat002 @nicusX. Using Database Parameters and Table Parameters we are now able to use lower/upper case for defining the Database and Table Name. By using the Show Tables/Show Databases, Describe table commands, we will show the original Flink Definition, even though in Glue UI it will be all in lower case

[Samrat002]

Thanks for incorporating the changes. I will review the pr in next couple of days .

[Samrat002]

Patch looks great. Thank you for making the change.
minor comments
Cheers,
Samrat

[melin]

Can we refer to Paimon's solution for integrating Glue?
https://github.com/apache/paimon/pull/4283/files#diff-534e1f558f31481abfd234e3b9c56c17a0988acf360a2264b0a090fa3a6aedd5

https://issues.apache.org/jira/browse/FLINK-38457

@FranMorilloAWS

[fmorillo7694]

@melin. To me knowledge Paimon integration with Glue Data Catalog is to store Paimon Tables in Glue, however it wouldnt work for storing Kinesis, MSK, and other streaming sources. They have different implementations.

[fmorillo7694]

Real AWS Glue integration test results (design rationale for originalName)

The latest head (b1a05d3) includes fixes driven by an integration test run against a real Glue Data Catalog (us-west-2), not Moto/Localstack. Posting the evidence here since it settles a design question a reviewer might reasonably raise: why store lowercase names plus an originalName parameter instead of writing mixed-case names directly?

Empirical facts about real Glue (reproducible via the AWS CLI)

1. Glue lowercases column names on CreateTable. Creating a table with columns userId / EventTime stores and returns userid / eventtime. Glue does not preserve mixed-case column names, so writing them as-is silently loses case.
2. Glue rejects column-level Parameters on partition columns with 400: Parameters not supported for partition columns. The originalName column-parameter mechanism therefore cannot be used for partition keys — their declared case is preserved in an order-preserving table-level parameter (flink.original-partition-keys) instead.

What case loss breaks (observed before the fix, same suite)

• DESCRIBE returned [userid, eventtime, username] for a table declared [userId, EventTime, userName]
• SELECT userId FROM t → Column 'userId' not found in any table; did you mean 'userid'?
• Partition ops failed to match: spec {eventDate=...} vs stored keys [eventdate]
• Streaming INSERT INTO / SELECT against catalog-registered Kinesis tables failed to plan, and JSON field names on the wire would be lowercased (silent data corruption for consumers)

Validation on the current head

25-step integration suite against real AWS, all passing:

• DDL: mixed-case databases, tables, and columns round-trip exactly (create / show / describe / drop); ALTER TABLE ... SET (now implemented via UpdateTable) with a case-preservation regression check; partitioned tables with ADD/SHOW/DROP PARTITION on mixed-case partition keys
• Streaming: unbounded job writing to and reading from a real Kinesis stream through source/sink tables registered in this catalog; JSON field names verified case-intact on the wire via GetRecords; PARTITIONED BY on a Kinesis table verified to set the record partition key from the (case-restored) column
• Unit tests: 126/126

One known behavior worth documenting: partition columns are appended after data columns on read-back (Glue models partition keys outside the storage descriptor, like Hive), so declaring partition columns last keeps the schema order stable across the round-trip.

[fmorillo7694]

@Samrat002 All open review threads are now addressed and resolved as of 002c0fb:

• Simplified GlueCatalog#close() (removed the retry logic, per @leekeiabstraction's finding that the SDK close() is best-effort)
• Removed the leftover comment in GlueCatalogFactory
• Initialized glueFunctions in GlueFunctionOperator#listGlueFunctions so it can never be null
• Added null checks on the GlueOperator constructor params
• Docs: clarified DataStream API usage via StreamTableEnvironment and linked the AWS credentials configuration guide

The branch is also synced with main (0 commits behind, including the Flink 2.1.2 / connector-parent 2.0.0 bump), module poms bumped to 6.1-SNAPSHOT, missing ASF license headers added, and the new spotless formatting applied. mvn verify passes locally (126/126 tests, spotless + RAT clean).

The CI workflow run on the new head is waiting for approval (action_required) — could you approve it and take another review pass when you get a chance? Thank you!

[fmorillo7694]

Test evidence: full suite executed against real AWS Glue

Following up on the review discussions about test coverage, the three commits just pushed (38fd97b, d600182, eae545c) extend the test pyramid and make it executable against the real service. Evidence below is from a single mvn verify invocation against real AWS Glue in eu-central-1 (screenshot to follow in the next comment).

What was run

IT_CASE_GLUE_CATALOG_USE_DEFAULT_CREDENTIALS=true IT_CASE_GLUE_CATALOG_REGION=eu-central-1 \
  mvn -pl flink-catalog-aws/flink-catalog-aws-glue,flink-connector-aws-e2e-tests/flink-catalog-aws-glue-e2e-tests \
  -am verify -Prun-aws-end-to-end-tests -Dsurefire.rerunFailingTestsCount=3

[WARNING] Tests run: 127, Failures: 0, Errors: 0, Skipped: 18, Flakes: 5   <- unit suite vs REAL Glue
[INFO] Tests run: 5, ... -- in ...glue.GlueCatalogMotoITCase               <- wire-level (moto, CI tier)
[INFO] Tests run: 3, ... -- in ...glue.GlueCatalogSqlMotoITCase            <- SQL path (moto, CI tier)
[INFO] Tests run: 2, ... -- in ...glue.test.GlueCatalogEndToEndITCase      <- e2e vs REAL Glue
[INFO] BUILD SUCCESS

• The 18 skips are fault-injection tests (simulated Glue server errors) — impossible against the real service by definition; they skip with that reason and run normally in CI's fake mode.
• The 5 flakes are Glue eventual-consistency transients (CreateDatabase → immediate read can briefly miss; DeleteDatabase is asynchronous); they passed on automatic rerun and are reported by surefire rather than hidden.
• CI behaviour is unchanged: without credentials the unit suite runs against the in-memory fake exactly as before (verified: 127/127, zero skips), and the e2e module is compile-only.

What the new tiers add

Tier	Tests	Runs in CI?	Backend
Unit suite (existing, now backend-switchable)	127	yes (fake)	fake or real Glue via GlueTestClientFactory
GlueCatalogMotoITCase — catalog API over the real SDK wire path	5	yes	moto (Testcontainers)
GlueCatalogSqlMotoITCase — CREATE CATALOG → SPI factory → planner → wire	3	yes	moto (Testcontainers)
GlueCatalogEndToEndITCase — full user path incl. a data round-trip (real Flink job between catalog-registered tables)	2	compile-only (-Prun-aws-end-to-end-tests + credentials)	real AWS Glue

The e2e run also covers the two behaviours only real Glue exhibits: column-name lowercasing with case restoration through the catalog, and dropDatabase (whose emptiness check requires the UDF API that emulators do not implement).

Bug found and fixed by the new coverage

The e2e data round-trip immediately surfaced a contract violation no DDL-only test could reach: GlueCatalog#getFunction threw a bare CatalogException when the database did not exist, but the planner probes getFunction on the current database for every SQL expression and only falls back to built-in functions on FunctionNotExistException — so any expression query failed SQL validation whenever the session's current database was absent from Glue. Fixed in 38fd97b with a regression test.

[fmorillo7694]

[fmorillo7694]

@Samrat002

[fmorillo7694]

@Samrat002 Good catches, both fixed in 649f4d9: removed the tracked .idea/ directory from the catalog module (the root .gitignore's */.idea/ pattern doesn't match nested modules, which is how it slipped in) and restored the root .idea/vcs.xml this branch had accidentally deleted. Also removed the module's META-INF/NOTICE — the module doesn't bundle any dependencies (its bundled-deps section was empty), so per repo convention only the bundling modules carry a hand-written NOTICE and the jar NOTICE comes from maven-remote-resources.