# Round-Trip an Apache Iceberg Table This example reads and writes an Apache Iceberg™ table on Confluent Manager for Apache Flink (CMF) from end to end. You create an object storage, configure CMF and build the Iceberg connector, then create a table, write three rows, and read them back. #### IMPORTANT Reading an Iceberg table back through Flink SQL requires CMF 2.4.2 or later. Writing requires a `cp-flink-sql` image that carries the connector classloader fix (1.19-cp11 or later). Without the image fix, the `INSERT` job fails; without CMF 2.4.2, the `SELECT` step fails to load the Iceberg catalog. ## Prerequisites Complete the [examples setup](setup.md#cmf-sql-examples-setup) first. It installs the MinIO object store and enables artifact management, which this example builds on. Iceberg needs extra CMF configuration and a dedicated compute pool, both covered below. The credentials (`console` / `console123`) and namespace (`flink`) match the setup. Substitute your own. ## Why Iceberg needs configuration on both sides Iceberg’s Hadoop catalog runs in two places, so both need to reach the object store. It runs inside the Flink cluster (the `INSERT` and `SELECT` jobs), and it runs inside CMF itself when CMF compiles a statement. Every statement that uses the table loads the catalog to plan the job, and planning the first one, the `INSERT`, writes the table’s first metadata file. You give the Flink side its credentials through the compute pool, and the CMF side through the configuration in [Configure CMF for Iceberg](#cmf-sql-example-iceberg-configure-cmf). ## Configure CMF for Iceberg CMF needs S3 access on its compile path, a larger artifact upload limit, and a wider class allowlist. First create a `core-site.xml` ConfigMap in CMF’s namespace so CMF’s Iceberg catalog can reach the object store (CMF is not a Flink process, so it reads Hadoop settings from `HADOOP_CONF_DIR` rather than from a pool’s `flink.hadoop.*` keys): ```yaml apiVersion: v1 kind: ConfigMap metadata: name: cmf-hadoop-conf namespace: flink data: core-site.xml: | fs.s3a.endpointhttp://minio.flink.svc.cluster.local:9000 fs.s3a.path.style.accesstrue fs.s3a.connection.ssl.enabledfalse fs.s3a.access.keyconsole fs.s3a.secret.keyconsole123 fs.s3a.aws.credentials.providerorg.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider fs.s3a.implorg.apache.hadoop.fs.s3a.S3AFileSystem ``` Apply it, then upgrade CMF with the following `iceberg-values.yaml`. These values extend the [examples setup](setup.md#cmf-sql-examples-setup) values with the Iceberg-specific settings, so this single upgrade is all the Iceberg example needs: it raises `maxUploadSize` above the ~112 MB connector JAR, adds the JDK packages the Iceberg connector loads at compile time to the class allowlist, and mounts the ConfigMap with the AWS environment the S3A credential chain reads. ```yaml cmf: artifacts: enabled: true basePath: s3://flink/cmf-artifacts maxUploadSize: 300MB configuration: s3.endpoint: http://minio.flink.svc.cluster.local:9000 s3.path.style.access: "true" s3.connection.ssl.enabled: "false" s3.access-key: console s3.secret-key: console123 sql: environmentCatalog: enabled: true examplesCatalog: enabled: true # The default allowlist plus the JDK packages a Hadoop/Iceberg connector # loads at compile time (org.w3c.dom, org.xml.sax, org.ietf.jgss, sun.misc). # Helm replaces this list, so it must repeat the defaults. classAllowlist: - 'java\..*' - 'javax\..*' - 'jdk\.internal\..*' - 'org\.apache\.flink\..*' - 'io\.confluent\.flink\..*' - 'org\.slf4j\..*' - 'org\.apache\.kafka\..*' - 'org\.w3c\.dom\..*' - 'org\.xml\.sax\..*' - 'org\.ietf\.jgss\..*' - 'sun\.misc\..*' extraEnv: - name: HADOOP_CONF_DIR value: /opt/hadoop-conf - name: AWS_ACCESS_KEY_ID value: console - name: AWS_SECRET_ACCESS_KEY value: console123 - name: AWS_REGION value: us-east-1 mountedVolumes: volumes: - name: hadoop-conf configMap: name: cmf-hadoop-conf volumeMounts: - name: hadoop-conf mountPath: /opt/hadoop-conf ``` `--reuse-values` keeps your other settings, but Helm replaces lists instead of merging them. The `extraEnv` and `mountedVolumes` lists in `iceberg-values.yaml` replace the ones in your release. If your release already sets either list, copy its entries into `iceberg-values.yaml` first. To see the current values, run `helm get values cmf --namespace flink`. ```bash kubectl apply -f cmf-hadoop-conf.yaml helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink --version "~2.4.0" \ --namespace flink \ --reuse-values \ --values iceberg-values.yaml ``` ## Build the connector JAR The connector is a fat JAR that bundles the Iceberg Flink runtime, an S3A filesystem, and a slim AWS SDK. Iceberg’s published `iceberg-flink-runtime` JAR carries neither `hadoop-aws` nor an AWS SDK, so an S3 warehouse needs those shaded in. The bundle also relocates Flink’s `HadoopUtils` class so the connector resolves Hadoop from its own JAR rather than CMF’s Hadoop-free host classpath. The versions are load-bearing: `iceberg-flink-runtime-1.19` 1.10.2 is the last Iceberg release that supports Flink 1.19, and the AWS SDK version must match the one `hadoop-aws` was built against. All dependencies are on Maven Central. `pom.xml`: ```xml 4.0.0 io.confluent.examples.flink iceberg-flink-1.19-s3-bundle 1.0.0 jar 11 UTF-8 1.10.2 3.4.1 1.19.1 2.24.6 software.amazon.awssdk bom ${awssdk.version} pom import org.apache.iceberg iceberg-flink-runtime-1.19 ${iceberg.version} org.apache.hadoop hadoop-client-api ${hadoop.version} org.apache.hadoop hadoop-client-runtime ${hadoop.version} org.apache.hadoop hadoop-aws ${hadoop.version} org.apache.hadoop hadoop-common org.apache.hadoop hadoop-client software.amazon.awssdk bundle software.amazon.awssdk s3 software.amazon.awssdk s3-transfer-manager software.amazon.awssdk apache-client software.amazon.awssdk sts org.apache.flink flink-hadoop-fs ${flink.version} iceberg-flink-1.19-s3-bundle-${project.version} org.apache.maven.plugins maven-shade-plugin 3.5.3 package shade false org.apache.flink:flink-hadoop-fs org/apache/flink/runtime/util/HadoopUtils*.class *:* META-INF/*.SF META-INF/*.DSA META-INF/*.RSA module-info.class META-INF/versions/*/module-info.class org.apache.flink.runtime.util.HadoopUtils shaded.flink.runtime.util.HadoopUtils ``` Build it: ```bash mvn clean package ``` The build writes `target/iceberg-flink-1.19-s3-bundle-1.0.0.jar` (about 112 MB). ## Create the compute pool Create a compute pool on the `cp-flink-sql:1.19-cp11` image whose configuration carries the Hadoop S3A credentials the Iceberg catalog uses on the Flink side, not only the Flink `s3.*` keys. Save this as `iceberg-pool.json`: ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "ComputePool", "metadata": { "name": "iceberg-pool" }, "spec": { "type": "DEDICATED", "clusterSpec": { "flinkVersion": "v1_19", "image": "confluentinc/cp-flink-sql:1.19-cp11", "flinkConfiguration": { "taskmanager.numberOfTaskSlots": "4", "execution.checkpointing.interval": "2000", "state.checkpoints.dir": "s3://flink/iceberg-pool/checkpoints", "state.savepoints.dir": "s3://flink/iceberg-pool/savepoints", "high-availability.storageDir": "s3://flink/iceberg-pool/recovery", "s3.endpoint": "http://minio.flink.svc.cluster.local:9000", "s3.path.style.access": "true", "s3.connection.ssl.enabled": "false", "s3.access-key": "console", "s3.secret-key": "console123", "flink.hadoop.fs.s3a.endpoint": "http://minio.flink.svc.cluster.local:9000", "flink.hadoop.fs.s3a.path.style.access": "true", "flink.hadoop.fs.s3a.connection.ssl.enabled": "false", "flink.hadoop.fs.s3a.access.key": "console", "flink.hadoop.fs.s3a.secret.key": "console123", "flink.hadoop.fs.s3a.aws.credentials.provider": "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider" }, "taskManager": { "resource": { "cpu": 0.5, "memory": "1024m" } }, "jobManager": { "resource": { "cpu": "500m", "memory": "1536m" } } } } } ``` ```bash confluent flink compute-pool create iceberg-pool.json \ --environment test \ --url http://cmf:8080 ``` ## Upload the connector artifact Label the JAR as a `connector` with the `iceberg` identifier, in `artifact.json`: ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "Artifact", "metadata": { "name": "iceberg-connector.jar", "labels": { "cmf.confluent.io/sql-artifact-type": "connector", "cmf.confluent.io/sql-artifact-identifier": "iceberg" } }, "spec": {} } ``` Upload it. The JAR is large, so allow for the upload time: ```bash curl -X POST http://cmf:8080/cmf/api/v1/environments/test/artifacts \ -F 'artifact=@artifact.json;type=application/json' \ -F 'file=@target/iceberg-flink-1.19-s3-bundle-1.0.0.jar' ``` The response reaches `status.phase` of `READY`. ## Create the table Create an Iceberg table in the environment catalog with a Hadoop catalog on the object store. `CREATE TABLE` checks that the connector is available and stores the options. It does not download the connector yet. ```bash confluent flink statement create ice-create \ --url http://cmf:8080 \ --environment test \ --compute-pool iceberg-pool \ --catalog _env_test \ --database default \ --sql "CREATE TABLE \`ice_orders\` (\`id\` BIGINT, \`data\` STRING) WITH ('connector' = 'iceberg', 'catalog-type' = 'hadoop', 'catalog-name' = 'hadoop_s3', 'warehouse' = 's3a://flink/iceberg-warehouse');" ``` The statement reaches `COMPLETED` with `sqlKind` of `CREATE_TABLE`. ## Write rows Insert three rows. This is the first statement to use the table, so CMF resolves the `iceberg` artifact and ships it to the Flink cluster. The bounded `INSERT` reaches `COMPLETED` when the rows are written. ```bash confluent flink statement create ice-insert \ --url http://cmf:8080 \ --environment test \ --compute-pool iceberg-pool \ --catalog _env_test \ --database default \ --sql "INSERT INTO \`ice_orders\` VALUES (1, 'a'), (2, 'b'), (3, 'c');" ``` Confirm the data landed in object storage. The Hadoop catalog lays the table out under `//`: ```bash kubectl -n flink port-forward svc/minio 9000:9000 & mc alias set minio http://localhost:9000 console console123 mc ls --recursive minio/flink/iceberg-warehouse/ ``` ```text default/ice_orders/data/00000-0-...-00001.parquet default/ice_orders/metadata/snap-...-1-....avro default/ice_orders/metadata/....-m0.avro default/ice_orders/metadata/v2.metadata.json default/ice_orders/metadata/version-hint.text ``` ## Read rows back Query the table. CMF resolves and ships the same artifact, and the job reads the current snapshot: ```bash confluent flink statement create ice-select \ --url http://cmf:8080 \ --environment test \ --compute-pool iceberg-pool \ --catalog _env_test \ --database default \ --sql "SELECT * FROM \`ice_orders\`;" ``` The query returns the three rows: ```text id data 1 a 2 b 3 c ``` ## Clean up Drop the table, then remove the pool and artifact: ```bash confluent flink statement create ice-drop \ --url http://cmf:8080 --environment test --compute-pool iceberg-pool \ --catalog _env_test --database default \ --sql "DROP TABLE IF EXISTS \`ice_orders\`;" curl -X DELETE http://cmf:8080/cmf/api/v1/environments/test/compute-pools/iceberg-pool curl -X DELETE http://cmf:8080/cmf/api/v1/environments/test/artifacts/iceberg-connector.jar ``` Dropping the table removes it from the environment catalog. Remove the Iceberg warehouse files from object storage to reclaim the space: ```bash mc rm --recursive --force minio/flink/iceberg-warehouse/ ``` ## Related content - [Custom Connectors and Formats in Confluent Manager for Apache Flink](../custom-connectors.md#cmf-sql-custom-connectors) - [Manage Artifacts in Confluent Manager for Apache Flink](../../../configure/artifacts.md#cmf-artifacts) - [Flink SQL Examples in Confluent Manager for Apache Flink](index.md#cmf-sql-examples)