You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "norrishuang (via GitHub)" <gi...@apache.org> on 2026/03/12 13:40:02 UTC
[PR] [FLINK-CDC][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
norrishuang opened a new pull request, #4314:
URL: https://github.com/apache/flink-cdc/pull/4314
## Summary
Add AWS Glue Catalog support for the Iceberg pipeline sink connector. Previously, the Iceberg pipeline only supported `hadoop` and `hive` catalog types. This PR enables users to use AWS Glue Data Catalog as the Iceberg catalog by setting `catalog.properties.type: glue`.
## Changes
### New Configuration Options
| Option | Type | Description |
|--------|------|-------------|
| `catalog.properties.type` | String | Now supports `glue` in addition to `hadoop` and `hive` |
| `catalog.properties.catalog-impl` | String | Custom catalog implementation class (e.g. `org.apache.iceberg.aws.glue.GlueCatalog`) |
| `catalog.properties.io-impl` | String | Custom FileIO implementation (e.g. `org.apache.iceberg.aws.s3.S3FileIO`) |
| `catalog.properties.glue.id` | String | Glue Catalog ID (AWS account ID) for cross-account access |
| `catalog.properties.glue.skip-archive` | Boolean | Skip archiving older table versions in Glue (default: true) |
| `catalog.properties.glue.skip-name-validation` | Boolean | Skip name validation for Glue catalog (default: false) |
| `catalog.properties.client.region` | String | AWS region for the Glue catalog client |
### Files Modified
- `IcebergDataSinkOptions.java` — Added Glue-related config options, updated `TYPE` and `WAREHOUSE` descriptions
- `IcebergDataSinkFactory.java` — Registered new optional config options
- `IcebergDataSinkFactoryTest.java` — Added 2 test cases for Glue catalog creation (via `type=glue` and `catalog-impl`)
- `pom.xml` — Dependency and shade plugin adjustments
## Usage Example
```yaml
sink:
type: iceberg
catalog.properties.type: glue
catalog.properties.warehouse: s3://my-bucket/warehouse/
catalog.properties.io-impl: org.apache.iceberg.aws.s3.S3FileIO
catalog.properties.client.region: us-east-1
```
## How It Works
Iceberg's CatalogUtil.buildIcebergCatalog() natively supports type=glue and automatically loads org.apache.iceberg.aws.glue.GlueCatalog. This PR exposes the necessary configuration options through the Flink CDC pipeline config layer and ensures the Glue-related catalog properties are correctly passed through via the catalog.properties.* prefix.
### Testing
Unit tests pass (6/6 in IcebergDataSinkFactoryTest)
Verified end-to-end on Amazon EMR (Flink 1.20, Iceberg 1.10.0-amzn) with MySQL CDC → Iceberg (Glue Catalog + S3)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#issuecomment-4053657742
Hi @norrishuang, would you like to update the related doc for this, and it should reflect which additional jars need to be added.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "norrishuang (via GitHub)" <gi...@apache.org>.
norrishuang commented on PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#issuecomment-4087990826
Updated docs: added `iceberg-flink-runtime-1.20` to the dependency table (required for all catalog types), and changed `catalog.properties.type` / `catalog.properties.catalog-impl` to conditionally required (either one must be set).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#discussion_r2957513998
##########
docs/content/docs/connectors/pipeline-connectors/iceberg.md:
##########
@@ -63,20 +65,60 @@ pipeline:
parallelism: 2
```
+### AWS Glue Catalog Example
+
+```yaml
+source:
+ type: mysql
+ name: MySQL Source
+ hostname: 127.0.0.1
+ port: 3306
+ username: admin
+ password: pass
+ tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
+ server-id: 5401-5404
+
+sink:
+ type: iceberg
+ name: Iceberg Sink
+ catalog.properties.type: glue
+ catalog.properties.warehouse: s3://my-bucket/warehouse
+ catalog.properties.io-impl: org.apache.iceberg.aws.s3.S3FileIO
+ catalog.properties.client.region: us-east-1
+ catalog.properties.glue.skip-archive: true
+
+pipeline:
+ name: MySQL to Iceberg via Glue Pipeline
+ parallelism: 2
+```
+
***Note:***
-If `catalog.properties.type` is hadoop, you need to configure the following dependencies manually, and pass it with `--jar` argument of Flink CDC CLI when submitting YAML pipeline jobs.
+Depending on the catalog type, you may need to add extra JARs manually and pass them with the `--jar` argument of Flink CDC CLI when submitting YAML pipeline jobs.
+
<div class="wy-table-responsive">
<table class="colwidths-auto docutils">
<thead>
<tr>
+ <th class="text-left">Catalog Type</th>
<th class="text-left">Dependency Item</th>
<th class="text-left">Description</th>
</tr>
</thead>
<tbody>
<tr>
- <td><a href="https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop-2-uber/2.8.3-10.0 "> org.apache.flink:flink-shaded-hadoop-2-uber:2.8.3-10.0</a></td>
- <td>Used for Hadoop dependencies.</td>
+ <td>hadoop</td>
+ <td><a href="https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop-2-uber/2.8.3-10.0">org.apache.flink:flink-shaded-hadoop-2-uber:2.8.3-10.0</a></td>
+ <td>Provides Hadoop filesystem dependencies.</td>
+ </tr>
+ <tr>
+ <td>glue</td>
+ <td><a href="https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-aws">org.apache.iceberg:iceberg-aws</a></td>
+ <td>Provides AWS Glue Catalog and S3 FileIO implementation.</td>
Review Comment:
The dependency note/table doesn’t mention that the connector itself no longer bundles Iceberg runtime classes (the module now declares `iceberg-flink-runtime-1.20` as `provided` and removes shading). Users running outside environments that pre-install Iceberg will hit `NoClassDefFoundError` unless they also add the Iceberg runtime JAR(s) to the job/cluster classpath. Please update this dependency table/note to explicitly include `org.apache.iceberg:iceberg-flink-runtime-1.20` (and clarify it’s required for all catalog types when not present in the runtime).
##########
docs/content/docs/connectors/pipeline-connectors/iceberg.md:
##########
@@ -115,21 +157,63 @@ Pipeline Connector Options
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Metastore of Iceberg catalog, supports <code>hadoop</code> and <code>hive</code>.</td>
+ <td>Metastore type of Iceberg catalog, supports <code>hadoop</code>, <code>hive</code>, and <code>glue</code>. Alternatively, you can use <code>catalog.properties.catalog-impl</code> to specify a custom catalog class directly.</td>
Review Comment:
In the options table, `catalog.properties.type` is marked as **required**, but the description says users can alternatively set `catalog.properties.catalog-impl` instead. Please adjust the “Required” column and/or wording to reflect the actual constraint (e.g., “either `catalog.properties.type` or `catalog.properties.catalog-impl` must be set”).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#issuecomment-4095217502
The failed case is related to the changes made here.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "norrishuang (via GitHub)" <gi...@apache.org>.
norrishuang commented on code in PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#discussion_r2929857536
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/pom.xml:
##########
@@ -35,6 +35,13 @@ limitations under the License.
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-flink-runtime-1.20</artifactId>
<version>${iceberg.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-aws</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>provided</scope>
Review Comment:
These two dependencies may conflict with existing dependencies in the runtime environment; it is recommended not to include them in the current package.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "norrishuang (via GitHub)" <gi...@apache.org>.
norrishuang commented on code in PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#discussion_r2929963064
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergDataSinkFactoryTest.java:
##########
@@ -111,6 +111,48 @@ void testPrefixRequireOption() {
Assertions.assertThat(dataSink).isInstanceOf(IcebergDataSink.class);
}
+ @Test
+ void testCreateGlueCatalogDataSink() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("iceberg", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(IcebergDataSinkFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put("catalog.properties.type", "glue")
+ .put("catalog.properties.warehouse", "s3://my-bucket/warehouse")
+ .put("catalog.properties.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
+ .put("catalog.properties.client.region", "us-east-1")
+ .put("catalog.properties.glue.skip-archive", "true")
+ .build());
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSink).isInstanceOf(IcebergDataSink.class);
+ }
+
+ @Test
+ void testCreateGlueCatalogWithCatalogImpl() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("iceberg", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(IcebergDataSinkFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put("catalog.properties.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
+ .put("catalog.properties.warehouse", "s3://my-bucket/warehouse")
+ .put("catalog.properties.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
+ .build());
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSink).isInstanceOf(IcebergDataSink.class);
+ }
Review Comment:
`iceberg-aws` is declared with provided scope (supplied by EMR at runtime), so GlueCatalog is not on the test classpath. Even if we added it as a test dependency, GlueCatalog.initialize() would try to connect to AWS Glue, which would fail in CI without valid AWS credentials.
These Glue tests are intentionally scoped to verify that the factory correctly parses Glue catalog configurations. Actual catalog construction and connectivity should be covered by integration tests with proper AWS access.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#discussion_r2929506068
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergDataSinkOptions.java:
##########
@@ -35,14 +35,60 @@ public class IcebergDataSinkOptions {
key("catalog.properties.type")
.stringType()
.noDefaultValue()
- .withDescription("Type of iceberg catalog, supports `hadoop` and `hive`.");
+ .withDescription(
+ "Type of iceberg catalog, supports `hadoop`, `hive` and `glue`.");
+
+ public static final ConfigOption<String> CATALOG_IMPL =
+ key("catalog.properties.catalog-impl")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Custom catalog implementation class. "
+ + "For AWS Glue catalog, use `org.apache.iceberg.aws.glue.GlueCatalog`.");
+
+ public static final ConfigOption<String> IO_IMPL =
+ key("catalog.properties.io-impl")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Custom FileIO implementation class. "
+ + "For AWS S3, use `org.apache.iceberg.aws.s3.S3FileIO`.");
+
+ public static final ConfigOption<String> GLUE_ID =
+ key("catalog.properties.glue.id")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The Glue catalog ID (AWS account ID). By default, the caller's AWS account ID is used.");
+
+ public static final ConfigOption<Boolean> GLUE_SKIP_ARCHIVE =
+ key("catalog.properties.glue.skip-archive")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to skip archiving older table versions in Glue. Default is true.");
+
+ public static final ConfigOption<Boolean> GLUE_SKIP_NAME_VALIDATION =
+ key("catalog.properties.glue.skip-name-validation")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to skip name validation for Glue catalog. Default is false.");
+
+ public static final ConfigOption<String> CLIENT_REGION =
+ key("catalog.properties.client.region")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The AWS region for the Glue catalog client.");
public static final ConfigOption<String> WAREHOUSE =
key("catalog.properties.warehouse")
.stringType()
.noDefaultValue()
.withDescription(
- "The warehouse root path of catalog, only usable when catalog.properties.type is `hadoop`.");
+ "The warehouse root path of catalog. "
+ + "For `hadoop` catalog, this is the local or HDFS path. "
+ + "For `glue` catalog, this is typically an S3 path like `s3://my-bucket/warehouse`.");
Review Comment:
`WAREHOUSE` description was updated for Hadoop/Glue, but it no longer mentions the Hive catalog case. Since `catalog.properties.warehouse` is also commonly used/required for HiveCatalog setups, consider updating the description to cover `hive` as well (or state that it applies to all catalog types, with examples for each).
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/pom.xml:
##########
@@ -291,15 +298,8 @@ limitations under the License.
<shadeTestJar>false</shadeTestJar>
<artifactSet>
<includes>
Review Comment:
The shade plugin configuration now has an empty `<artifactSet><includes>`, which typically results in shading nothing (or at best makes the output dependent on plugin defaults). This looks accidental, especially since other pipeline connectors explicitly include their external dependencies for self-contained connector JARs. Consider restoring explicit includes (e.g., `org.apache.iceberg:*` and `org.apache.iceberg:iceberg-aws`) and, if Iceberg still pulls Kafka clients, restoring the Kafka relocation to avoid dependency conflicts.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/pom.xml:
##########
@@ -35,6 +35,13 @@ limitations under the License.
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-flink-runtime-1.20</artifactId>
<version>${iceberg.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+ <artifactId>iceberg-aws</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>provided</scope>
Review Comment:
Changing Iceberg dependencies to scope `provided` means the published connector JAR will no longer carry `iceberg-flink-runtime-1.20` / `iceberg-aws` at runtime (and Maven Shade won’t include provided deps by default). Unless the distribution/runtime explicitly supplies these artifacts, this will cause `ClassNotFoundException` when the sink tries to build the catalog (especially for Glue). Consider keeping these as non-provided (and shading them), or document/ensure that users must add the Iceberg runtime + iceberg-aws (and AWS SDK) to the classpath.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/IcebergDataSinkFactoryTest.java:
##########
@@ -111,6 +111,48 @@ void testPrefixRequireOption() {
Assertions.assertThat(dataSink).isInstanceOf(IcebergDataSink.class);
}
+ @Test
+ void testCreateGlueCatalogDataSink() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("iceberg", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(IcebergDataSinkFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put("catalog.properties.type", "glue")
+ .put("catalog.properties.warehouse", "s3://my-bucket/warehouse")
+ .put("catalog.properties.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
+ .put("catalog.properties.client.region", "us-east-1")
+ .put("catalog.properties.glue.skip-archive", "true")
+ .build());
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSink).isInstanceOf(IcebergDataSink.class);
+ }
+
+ @Test
+ void testCreateGlueCatalogWithCatalogImpl() {
+ DataSinkFactory sinkFactory =
+ FactoryDiscoveryUtils.getFactoryByIdentifier("iceberg", DataSinkFactory.class);
+ Assertions.assertThat(sinkFactory).isInstanceOf(IcebergDataSinkFactory.class);
+
+ Configuration conf =
+ Configuration.fromMap(
+ ImmutableMap.<String, String>builder()
+ .put("catalog.properties.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog")
+ .put("catalog.properties.warehouse", "s3://my-bucket/warehouse")
+ .put("catalog.properties.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
+ .build());
+ DataSink dataSink =
+ sinkFactory.createDataSink(
+ new FactoryHelper.DefaultContext(
+ conf, conf, Thread.currentThread().getContextClassLoader()));
+ Assertions.assertThat(dataSink).isInstanceOf(IcebergDataSink.class);
+ }
Review Comment:
These new tests only assert that `createDataSink(...)` returns an `IcebergDataSink`, but `IcebergDataSinkFactory` doesn’t build the Iceberg catalog; the actual `CatalogUtil.buildIcebergCatalog(...)` happens later (e.g., in `IcebergWriter`/`IcebergCommitter`/`IcebergMetadataApplier`). As a result, the tests won’t catch missing Glue classes (e.g., if `iceberg-aws` isn’t on the runtime classpath) or invalid Glue-specific properties. Consider extending the test to actually invoke `CatalogUtil.buildIcebergCatalog` using the `catalogOptions` produced by the factory and assert the resulting catalog implementation (without making AWS calls).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39245][Iceberg] Add AWS Glue Catalog support for Iceberg pipeline connector [flink-cdc]
Posted by "norrishuang (via GitHub)" <gi...@apache.org>.
norrishuang commented on PR #4314:
URL: https://github.com/apache/flink-cdc/pull/4314#issuecomment-4053791840
> Hi @norrishuang, would you like to update the related doc for this, and it should reflect which additional jars need to be added.
The doc was updated.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org