qzyu999 commented on code in PR #3424: URL: https://github.com/apache/fluss/pull/3424#discussion_r3479374612
########## website/docs/streaming-lakehouse/integrate-data-lakes/catalogs/glue.md: ########## @@ -0,0 +1,411 @@ +--- +title: AWS Glue +sidebar_position: 2 +--- + +# AWS Glue + +## Introduction + +[AWS Glue](https://aws.amazon.com/glue/) is a serverless data integration service that includes a central metadata repository known as the AWS Glue Data Catalog. The Glue Data Catalog is compatible with Apache Iceberg, making it a convenient option for managing Iceberg table metadata on AWS. + +This guide explains how to configure Fluss to use AWS Glue as its Iceberg catalog. For general Iceberg integration details (table mapping, data types, limitations), see [Iceberg](../formats/iceberg.md). + +## How It Works + +When Fluss is configured with AWS Glue as its Iceberg catalog: + +1. **Data ingestion**: Applications write data to Fluss tables using the Fluss client (Java/Python), Flink SQL, or any Kafka-compatible producer. Fluss stores this data in its real-time log (similar to Kafka topics). +2. **Tiering to Iceberg**: A separate Flink job (the [tiering service](maintenance/tiered-storage/lakehouse-storage.md#start-the-datalake-tiering-service)) periodically reads accumulated data from Fluss, converts it to Parquet format, writes the files to S3, and commits an Iceberg snapshot to the Glue Data Catalog. +3. **Query via Athena/Spark/Trino**: Any Iceberg-compatible engine can discover and query the tiered tables through AWS Glue — no additional configuration needed. + +``` +┌─ Data Sources ─────────┐ ┌─ Fluss Cluster ──────┐ ┌─ AWS ────────────────────┐ +│ │ │ │ │ │ +│ App (Fluss client) ─┼────▶│ Coordinator │ │ Glue Data Catalog │ +│ Flink CDC job ─┼────▶│ TabletServer(s) │ │ (table metadata) │ +│ Kafka producer ─┼────▶│ ZooKeeper │ │ │ +│ │ │ │ │ S3 Bucket │ +└─────────────────────────┘ └──────────┬───────────┘ │ (Parquet data files) │ + │ │ │ + ┌──────────▼───────────┐ │ Athena / Spark / Trino │ + │ Flink Tiering Job ├────▶│ (query engine) │ + │ (reads Fluss log, │ │ │ + │ writes Iceberg) │ └──────────────────────────┘ + └──────────────────────┘ +``` + +> **KEY CONCEPT**: Flink is used here for the **tiering service** (Fluss → Iceberg/S3/Glue) and optionally for data ingestion via SQL. But Flink is NOT the only way to write data to Fluss. Applications can write directly using the Fluss client library (Java, Python) or any Kafka-compatible producer. The tiering service is what bridges Fluss's real-time log to the Glue Data Catalog. + +## Prerequisites + +### Java Version + +Fluss 0.9.x requires **Java 17 or later** for the Tablet Server. Java 11 will fail with `NoSuchMethodError: MappedByteBuffer.duplicate()`. + +### AWS IAM Permissions + +The processes running Fluss servers and the Flink tiering service must have IAM permissions for both Glue and S3. Below is a minimal IAM policy: + +```json +{ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": [ + "glue:CreateDatabase", + "glue:GetDatabase", + "glue:GetDatabases", + "glue:DeleteDatabase", + "glue:CreateTable", + "glue:GetTable", + "glue:GetTables", + "glue:UpdateTable", + "glue:DeleteTable" + ], + "Resource": [ + "arn:aws:glue:<region>:<account-id>:catalog", + "arn:aws:glue:<region>:<account-id>:database/*", + "arn:aws:glue:<region>:<account-id>:table/*" + ] + }, + { + "Effect": "Allow", + "Action": [ + "s3:GetObject", + "s3:PutObject", + "s3:DeleteObject", + "s3:ListBucket" + ], + "Resource": [ + "arn:aws:s3:::<your-bucket>", + "arn:aws:s3:::<your-bucket>/*" + ] + } + ] +} +``` + +> **NOTE**: If your account uses **AWS Lake Formation**, the IAM role must also have Lake Formation permissions (Create Table, Describe, Alter, Insert, Select, Delete, Drop) on the target database. Standard IAM `glue:*` permissions alone are not sufficient when Lake Formation governance is enabled. + +### Prepare Required JARs + +Fluss bundles `iceberg-core` but does **not** bundle the Glue catalog implementation or the AWS SDK. You must supply additional JARs. + +#### For Fluss Servers (Coordinator & Tablet Servers) + +Place the following JARs in the `${FLUSS_HOME}/plugins/iceberg/` directory: + +- **Iceberg AWS Bundle**: [iceberg-aws-bundle-1.10.1.jar](https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-aws-bundle/1.10.1/iceberg-aws-bundle-1.10.1.jar) +- **Iceberg AWS**: [iceberg-aws-1.10.1.jar](https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-aws/1.10.1/iceberg-aws-1.10.1.jar) +- **Failsafe**: [failsafe-3.3.2.jar](https://repo1.maven.org/maven2/dev/failsafe/failsafe/3.3.2/failsafe-3.3.2.jar) — required by `iceberg-aws` for Glue API retry logic + +Both Iceberg JARs are required. The bundle provides AWS SDK v2 dependencies, while `iceberg-aws` provides the `GlueCatalog` class that Iceberg loads via reflection. The plugin classloader cannot find `GlueCatalog` from the bundle alone. The `failsafe` library is a transitive dependency of `iceberg-aws` that is not bundled. + +> **NOTE**: You need **all three** JARs. Using only the bundle will result in `ClassNotFoundException: org.apache.iceberg.aws.glue.GlueCatalog`. Missing `failsafe` will cause `NoClassDefFoundError: dev/failsafe/FailsafeException`. + +Additionally, you need the Fluss S3 filesystem plugin for remote storage access. Place [fluss-fs-s3-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-fs-s3/$FLUSS_VERSION$/fluss-fs-s3-$FLUSS_VERSION$.jar) in `${FLUSS_HOME}/plugins/s3/`. See [S3 Dependencies](../../../maintenance/filesystems/s3.md#dependencies) for details. + +> **TIP**: The Fluss binary distribution already includes `fluss-lake-iceberg-$FLUSS_VERSION$.jar` in `plugins/iceberg/`. You do not need to download it separately — only add the three JARs above (iceberg-aws-bundle, iceberg-aws, failsafe). + +#### For the Flink Tiering Service + +Place the following JARs in `${FLINK_HOME}/lib`: + +1. **Iceberg AWS Bundle**: `iceberg-aws-bundle-1.10.1.jar` (same as above) +2. **Iceberg AWS**: `iceberg-aws-1.10.1.jar` (same as above — provides `GlueCatalog` class) +3. **Failsafe**: `failsafe-3.3.2.jar` (same as above — Glue retry logic) +4. **Fluss Flink Connector**: [fluss-flink-1.20-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-flink-1.20/$FLUSS_VERSION$/fluss-flink-1.20-$FLUSS_VERSION$.jar) (pick the version matching your Flink runtime) +5. **Fluss Lake Iceberg**: [fluss-lake-iceberg-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-lake-iceberg/$FLUSS_VERSION$/fluss-lake-iceberg-$FLUSS_VERSION$.jar) +6. **Fluss Flink Tiering**: [fluss-flink-tiering-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-flink-tiering/$FLUSS_VERSION$/fluss-flink-tiering-$FLUSS_VERSION$.jar) — the tiering job JAR itself +7. **Fluss S3 Filesystem**: [fluss-fs-s3-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-fs-s3/$FLUSS_VERSION$/fluss-fs-s3-$FLUSS_VERSION$.jar) (if S3 is used as Fluss remote storage) +8. **Hadoop Client**: [hadoop-client-api-3.3.6.jar](https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-client-api/3.3.6/hadoop-client-api-3.3.6.jar) and [hadoop-client-runtime-3.3.6.jar](https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-client-runtime/3.3.6/hadoop-client-runtime-3.3.6.jar) — required by the Iceberg Parquet writer + +> **NOTE**: Despite the Glue catalog itself not requiring Hadoop, the **tiering service** needs Hadoop classes (`org.apache.hadoop.conf.Configuration`) for writing Parquet files via Iceberg. Use the `hadoop-client-api` and `hadoop-client-runtime` JARs (not the full Hadoop distribution or `flink-shaded-hadoop-2-uber`) to avoid classpath conflicts with Flink's bundled Avro version. Using the uber JAR causes `NoSuchMethodError: LogicalTypes.timestampNanos()`. + +> **WARNING**: Do **not** add S3 credentials to `flink-conf.yaml` by appending key-value pairs — this can break Flink's memory configuration parser. Instead, use environment variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN`, `AWS_REGION`) or a `HADOOP_CONF_DIR` with a `core-site.xml` file. + +## Configure Fluss with AWS Glue + +### Cluster Configuration + +Add the following to your `server.yaml`: + +```yaml +datalake.format: iceberg +datalake.iceberg.type: glue +datalake.iceberg.warehouse: s3://<your-bucket>/<warehouse-path> +``` + +Fluss strips the `datalake.iceberg.` prefix and passes the remaining properties directly to Iceberg's Glue catalog. The properties above become `type=glue` and `warehouse=s3://...` when initializing the catalog. + +You can pass any [Iceberg AWS catalog property](https://iceberg.apache.org/docs/1.10.1/aws/#glue-catalog) using the same prefix. Common additional properties: + +```yaml +# Specify the AWS region (required if not using default region from credentials chain) +datalake.iceberg.client.region: us-east-1 + +# Use S3FileIO explicitly (Glue catalog defaults to this when warehouse is s3://) +datalake.iceberg.io-impl: org.apache.iceberg.aws.s3.S3FileIO + +# Optional: restrict catalog to a specific Glue database prefix +datalake.iceberg.glue.catalog-id: <aws-account-id> +``` + +#### Authentication + +**IAM Role (Recommended)**: If running on AWS (EKS, ECS, EC2) with an attached IAM role, no credential configuration is needed for the **Glue catalog connection**. The Iceberg AWS SDK uses the [default credentials provider chain](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials-chain.html) automatically. + +However, the **Fluss S3 filesystem plugin** (used for `remote.data.dir`) has a known limitation: it uses a custom delegation token mechanism that does not support IMDSv2 or ECS task role credentials. This affects: Review Comment: I validated the `glue.md` documentation end-to-end on AWS (ECS Fargate) with Fluss 0.9.1-incubating, Flink 1.20.1, Iceberg 1.10.1, and AWS Glue. This is a summary of findings. ## What Works ✅ The following are **validated and working** on ECS Fargate with the `s3.assumed.role.arn` credential approach: 1. **Fluss cluster starts** — Coordinator + TabletServer + ZooKeeper 2. **S3 filesystem plugin initializes** — "Using default AWS credential chain with AssumeRole" logged 3. **Glue catalog connection** — Iceberg GlueCatalog loads via `S3FileIO`, connects to Glue 4. **Table creation** — Flink SQL `CREATE TABLE ... WITH ('table.datalake.enabled'='true')` succeeds 5. **Table registered in Glue** — confirmed via `aws glue get-table` 6. **Data insertion** — Flink SQL INSERT succeeds 7. **Tiering job starts** — `flink run -d fluss-flink-tiering-0.9.1-incubating.jar` submits and runs (RUNNING state confirmed) 8. **Tiering reads data from TabletServer** — reaches the commit phase ## What Fails ❌ **Tiering snapshot commit fails** with a misleading error: ``` java.io.IOException: Fail to prepare commit table lake snapshot for db.table to Fluss. Caused by: java.lang.IllegalStateException: Field 'table_id' is not set ``` This is NOT a protobuf serialization issue. It's **two bugs** working together: ## Bug 1: Error masking in `CoordinatorService.prepareLakeTableSnapshot()` **File:** `fluss-server/.../CoordinatorService.java` (the handler for `PrepareLakeTableSnapshot` RPC) When `storeLakeTableOffsetsFile()` throws an exception, the catch block: - ✅ Sets `error_code` and `error_message` in the response - ❌ Does NOT set `table_id` in the response - ❌ Does NOT log the exception ```java // Server side (v0.9.1-incubating, exact released code): try { long tableId = bucketOffsets.getTableId(); // ... storeLakeTableOffsetsFile() throws here ... pbPrepareLakeTableRespForTable.setTableId(tableId); // ← only on success pbPrepareLakeTableRespForTable.setLakeTableOffsetsPath(fsPath.toString()); } catch (Exception e) { Errors error = ApiError.fromThrowable(e).error(); pbPrepareLakeTableRespForTable.setError(error.code(), error.message()); // ← table_id NOT SET // ← exception NOT LOGGED } ``` ## Bug 2: Client reads `table_id` before checking for errors **File:** `fluss-flink/.../FlussTableLakeSnapshotCommitter.java` ```java // Client side (v0.9.1-incubating, exact released code): prepareResp = pbPrepareLakeTableRespForTables.get(0); checkState( prepareResp.getTableId() == tableId, // ← CRASHES HERE (table_id not set in error response) ...); if (prepareResp.hasErrorCode()) { // ← NEVER REACHED throw ApiError.fromErrorMessage(prepareResp).exception(); // actual error would surface here } ``` **Result:** The real error from `storeLakeTableOffsetsFile()` is permanently lost. The user sees `"Field 'table_id' is not set"` which provides zero diagnostic value. The tiering service retries indefinitely. ## The Underlying Trigger: `s3.assumed.role.arn` doesn't pass credentials to Hadoop S3A The reason `storeLakeTableOffsetsFile()` fails is that it writes to the `remote.data.dir` (S3) using Hadoop S3A via the Fluss FileSystem abstraction. Looking at `S3FileSystemPlugin.setCredentialProvider()`: When `s3.assumed.role.arn` is set (without static keys), the plugin: - Logs "Using default AWS credential chain with AssumeRole" - Skips the delegation token credential provider setup - Does NOT set any explicit Hadoop S3A credentials (`fs.s3a.access.key`, etc.) - Relies on Hadoop's default credential provider chain On ECS Fargate, Hadoop's default chain should discover ECS container credentials. **But in practice, the `storeLakeTableOffsetsFile()` write fails** — likely because the bundled Hadoop version doesn't properly support the ECS container credential endpoint, or there's a classpath issue. **The previous approach (injecting temp credentials from the ECS metadata endpoint as `s3.access.key`/`s3.secret.key`/`s3.session.token`) DID work** because those get mapped directly to `fs.s3a.access.key`/`fs.s3a.secret.key`/`fs.s3a.session.token` — Hadoop S3A uses them for all file operations. ## Summary | Credential Approach | S3 Init | Table Creation (Glue) | Tiering (S3 write) | |---|---|---|---| | `s3.access.key` + `s3.secret.key` + `s3.session.token` (temp creds from ECS metadata) | ✅ | ✅ | ✅ (confirmed in earlier POC) | | `s3.assumed.role.arn` only | ✅ | ✅ | ❌ (`storeLakeTableOffsetsFile` fails silently) | ## Questions 1. The error handling in `CoordinatorService.prepareLakeTableSnapshot()` swallows exceptions — when `storeLakeTableOffsetsFile()` fails, the exception isn't logged and `table_id` isn't included in the error response. The client then crashes on a missing field before it ever reads the actual error. Can we get the exception logged and `table_id` set in the error path? Without this, any S3/filesystem failure in tiering is completely undebuggable. 2. In `FlussTableLakeSnapshotCommitter.prepareLakeSnapshot()`, the code calls `getTableId()` before checking `hasErrorCode()`. Since the error response doesn't include `table_id` (see above), this always throws `IllegalStateException` and hides the real error message. Flipping the check order would at least surface what's actually failing. 3. This is the main blocker for us: when we set `s3.assumed.role.arn` in `server.yaml`, the S3 filesystem plugin only uses it for delegation token generation — it does NOT configure Hadoop S3A (which actually writes files to S3) to use that role. So `storeLakeTableOffsetsFile()` tries to write to S3 with no credentials. On ECS Fargate/EKS, we'd expect `s3.assumed.role.arn` to mean "use this role for all S3 operations," not just for one internal subsystem. Is there a plan to make this work end-to-end? For now we're falling back to injecting `s3.access.key`/`s3.secret.key`/`s3.session.token` which does work but requires fetching temp creds from the ECS metadata endpoint at startup. ## Environment - Fluss 0.9.1-incubating (binary from apache.org) - `fluss-flink-tiering-0.9.1-incubating.jar` (from Maven Central) - Flink 1.20.1 - Iceberg 1.10.1 + iceberg-aws-bundle + iceberg-aws + failsafe - ECS Fargate - Hadoop: `hadoop-client-api-3.3.6` + `hadoop-client-runtime-3.3.6` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
