Skip to main content

​
Overview

Apache Iceberg Sink Connector continuously writes business records from Kafka topics to Iceberg tables, connecting real-time event streams with historical queries, offline analytics, and audit workflows in a data lake. Record values map to table fields, and written data files become visible to queries after they are committed through table snapshots; Kafka record keys do not automatically become table columns. A connector can write to one table, multiple fixed tables, or target tables selected by a record field. Without a routing field, each record is written to all fixed target tables; with routing configured, different categories of business events can be stored separately. It also supports automatic table creation and supported schema evolution, making it suitable for continuously appended order events, activity logs, and change histories. The built-in write path uses append semantics and does not automatically merge current state by business primary key. The connector uses Kafka transactions, a control topic, and Iceberg snapshot checkpoints to coordinate writes and recovery, providing exactly-once recovery semantics subject to prerequisites. These semantics require a compatible transaction protocol and Connect runtime, alignment with the actual data consumer group, a catalog that supports atomic snapshot commits, and retention of the control events, consumer group offsets, data files, and snapshot history needed for recovery. They do not imply business-key deduplication or provide atomic commits across multiple tables.

​
Prerequisites

  • Prepare an available Iceberg catalog, along with the client dependencies, authentication, and storage permissions required by the selected catalog and FileIO.
  • Automatic table creation is disabled by default. Pre-create the target tables that will receive records, and ensure that record fields are compatible with the target table schemas; a missing target actually selected by static routing causes the task to fail, whereas dynamic routing skips writes to missing tables.
  • When automatic table creation is enabled, the catalog must permit table creation, the target namespace must be available or its creation permitted, and the first record must allow a nonempty table schema to be inferred.
  • Prepare the control topic or allow broker-side automatic creation; internal clients must have the permissions required for control topic reads and writes, transactional IDs, and related consumer group operations.
  • Internal Kafka clients must connect to the Kafka cluster hosting the data consumer group so that control events and input offsets can be committed in the same Kafka transaction.
  • Kafka must support the transaction protocol that includes consumer group metadata, and the Connect runtime must be compatible with the way the connector obtains the underlying consumer.

​
License

Uses Apache License 2.0.

​
Quick Start

Prepare a Connect cluster, Kafka, an Iceberg REST catalog, and a target table in advance. Confirm network connectivity and access permissions for the catalog, storage, and control topic. For cluster and connector management, see Manage Connectors. The following configuration appends JSON objects without schema envelopes to an existing table.
Replace the input topic, fully qualified table name, REST catalog URI, and Kafka addresses. This example uses the default control topic control-iceberg, which must be available beforehand. Add REST catalog and storage access settings under iceberg.catalog.* as required by your deployment, and add the security properties required by internal Kafka clients under iceberg.kafka.*. Do not expose long-lived credentials in the configuration. Input values must be JSON objects compatible with the existing table fields; this example is not suitable for directly reading Debezium envelopes to maintain current state. By default, a table commit is initiated every 5 minutes. Consumer offsets may advance before table snapshots do, so a decrease in consumer lag does not mean that all data is already visible to queries.

​
Configuration

The following settings apply to append writes. Registered configuration defaults and runtime fallbacks are described separately. Dynamic properties do not have individually registered, uniform defaults or importance levels; validation is performed by the reading logic or the corresponding component. In per-table properties, <table-name> is the fully qualified target table name, including its namespace. For prefixed properties, N/A in the default and importance fields means these are not individually defined; it does not mean every delegated property lacks a default. Specific defaults and constraints are determined by the corresponding component. Per-table properties retain their actual read-time fallbacks.

​
Input and Tasks

​
connector.class

Selects the connector implementation class.
  • Type: string
  • Default: None
  • Importance: High
  • Required: Yes
  • Valid Values / Notes: Use org.apache.iceberg.connect.IcebergSinkConnector.

​
tasks.max

Sets the maximum number of tasks.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid Values / Notes: At least 1. Effective parallelism is limited by input partition assignment; tasks are not assigned one per table, and global ordering across partitions is not guaranteed.

​
topics

Specifies the list of input Kafka topics.
  • Type: list
  • Default: Empty list []
  • Importance: High
  • Valid Values / Notes: Comma-separated; choose either this setting or a nonempty topics.regex. This is not a target table list or the control topic.

​
topics.regex

Subscribes to input topics using a regular expression.
  • Type: string
  • Default: Empty string
  • Importance: High
  • Valid Values / Notes: A valid Java regular expression; choose either this setting or nonempty topics.

​
key.converter

Converts Kafka message keys to Connect data.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid Values / Notes: Inherits the worker converter when unset; an explicitly specified class must implement Converter. Message keys are not automatically written to table fields.

​
value.converter

Converts Kafka message values to Connect data.
  • Type: class
  • Default: null
  • Importance: Low
  • Valid Values / Notes: Inherits the worker converter when unset; there is no uniform JSON default. After conversion and SMT processing, a non-null top-level value must be a Struct or Map. Converter subproperties must match the actual input encoding.

​
consumer.override.*

Overrides properties of the Connect task’s data consumer. This property set has no uniform default; individual property types and defaults are determined by the Kafka Consumer.
  • Type: Pass-through Kafka Consumer properties
  • Default: N/A
  • Importance: N/A
  • Valid Values / Notes: Must be permitted by the worker’s client override policy. Settings take precedence in this order, from lowest to highest: Connect base values, worker consumer.*, and connector consumer.override.*. When overriding group.id, also update iceberg.connect.group-id to match. These properties do not configure internal control clients. auto.offset.reset only affects cases with no valid offset and cannot be used as a lossless recovery mechanism.

​
Catalog and Storage

​
iceberg.catalog

Sets the catalog instance name, not its implementation type.
  • Type: string
  • Default: iceberg
  • Importance: Medium
  • Valid Values / Notes: A nonempty set of iceberg.catalog.* properties is required even when using the default name.

​
iceberg.catalog.*

Passes properties, with the prefix removed, to the Iceberg catalog, including implementation selection, service addresses, authentication, and FileIO configuration. This property set has no uniform default; specific defaults are determined by the catalog and FileIO.
  • Type: Pass-through string properties
  • Default: N/A
  • Importance: N/A
  • Required: Yes
  • Valid Values / Notes: type and catalog-impl are mutually exclusive. type is case-insensitive and accepts hive, hadoop, rest, glue, nessie, or jdbc; actual availability depends on installed implementations and client dependencies. When neither is set, the underlying implementation falls back to Hive, but this is not a declaration of the connector’s default type. Whether uri, warehouse, io-impl, and other properties are required depends on the selected implementation.

​
iceberg.hadoop-conf-dir

Specifies the Hadoop XML configuration directory.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid Values / Notes: The directory must be accessible to the worker. It reads core-site.xml, hdfs-site.xml, and hive-site.xml from this directory and requires available Hadoop classes. iceberg.hadoop.* overrides corresponding properties after the XML files are loaded.

​
iceberg.hadoop.*

Sets properties, with the prefix removed, in the Hadoop Configuration. This property set has no uniform default; specific defaults are determined by Hadoop.
  • Type: Pass-through string properties
  • Default: N/A
  • Importance: N/A
  • Valid Values / Notes: Requires Hadoop classes. This is not a prefix for Kafka client or catalog properties.

​
Target Tables and Routing

​
iceberg.tables

Sets the static target table list.
  • Type: list
  • Default: null
  • Importance: High
  • Valid Values / Notes: Required in static mode; use comma-separated, fully qualified table names. Mutually exclusive with iceberg.tables.dynamic-enabled=true. An empty list provides no write targets. Without a routing field, each record is written to all static target tables.

​
iceberg.tables.dynamic-enabled

Enables dynamic routing that selects a target table from a record field.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: When true, remove iceberg.tables and set iceberg.tables.route-field. The routing value is converted to lowercase and used as the fully qualified table name; per-table routing regular expressions are not used. Null values are skipped, and invalid table identifiers may cause failure.

​
iceberg.tables.route-field

Specifies the routing field in the record value.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid Values / Notes: Required in dynamic mode; supports dot-separated paths through nested Struct or Map fields. In static mode, the field value is matched independently against each table’s route-regex. If the field is missing or null, no table receives the record.

​
iceberg.table.<table-name>.route-regex

Sets the routing match condition for a single static table.
  • Type: string
  • Default: Not registered; falls back to null when read
  • Importance: N/A
  • Valid Values / Notes: When a routing field is configured in static mode, provide a regular expression for each table that should receive data. Java regular expressions match the entire string; overlapping expressions write the record to multiple tables. Tables without a matching expression do not receive records. This setting is ignored in dynamic mode.

​
Automatic Table Creation and Schema

​
iceberg.tables.auto-create-enabled

Allows missing target tables to be created.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: Infers the table schema from the first record’s schema or value and requires table creation permissions. An empty object cannot be used to infer a schema. When disabled, a missing target actually selected by static routing throws NoSuchTableException and puts the task in the FAILED state; only dynamic routing skips writes for this missing-table exception. Dynamic missing-table skips do not apply to invalid table identifiers, permission errors, or other catalog exceptions.

​
iceberg.tables.evolve-schema-enabled

Allows supported table schema updates.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: Can add new fields whose types can be inferred. A Struct with a schema also supports promotion from int to long and from float to double, and relaxation of required fields to optional. Schemaless data does not have the same promotion path for existing fields. When disabled, extra fields are ignored. Arbitrary renaming, column deletion, or type migration is not supported.

​
iceberg.tables.schema-force-optional

Makes created or evolved fields optional.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: Also applies to container value types that are created or evolved; it does not rewrite the optionality of all fields in an existing table. By default, input schema optionality is respected.

​
iceberg.tables.schema-case-insensitive

Controls whether table field lookup is case-insensitive.
  • Type: boolean
  • Default: false
  • Importance: Medium
  • Valid Values / Notes: Does not rename fields. When the table has a name mapping, the mapping-based lookup path is used; this switch does not unconditionally override the mapping.

​
iceberg.tables.default-partition-by

Sets the default partition rules for automatic table creation.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid Values / Notes: Applies only to newly created tables and does not change existing table partitioning. Supports fields themselves and the year, month, day, hour, bucket, and truncate transforms. Separate multiple rules with commas; commas inside transform parentheses are preserved. For bucket and truncate, the argument order is column name, then width. Parsing or construction exceptions are logged as errors, and creation falls back to an unpartitioned table; configuration alone does not establish that partitioning took effect.

​
iceberg.table.<table-name>.partition-by

Overrides automatic table creation partition rules for a single table.
  • Type: string
  • Default: Not registered; falls back to iceberg.tables.default-partition-by when read
  • Importance: N/A
  • Valid Values / Notes: Used only for automatic table creation. An explicit empty string creates an unpartitioned table. Syntax and exception fallback behavior are the same as for the global partition rules.

​
iceberg.tables.auto-create-props.*

Provides Iceberg table properties for automatically created tables. This property set has no uniform default; specific defaults are determined by the Iceberg table and catalog.
  • Type: Pass-through Iceberg table properties
  • Default: N/A
  • Importance: N/A
  • Valid Values / Notes: Passed to the table creation operation with the prefix removed; does not update existing table properties.

​
iceberg.tables.write-props.*

Overrides Iceberg table properties used when initializing writers, such as file format and target file size. This property set has no uniform default; specific defaults are determined by Iceberg table properties.
  • Type: Pass-through Iceberg table properties
  • Default: N/A
  • Importance: N/A
  • Valid Values / Notes: Overrides writer-local properties without persistently modifying table properties. The target file size is not guaranteed to be reached on every commit.

​
Identifier Columns and Commit Branches

​
iceberg.tables.default-id-columns

Sets the default identifier column list.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid Values / Notes: Comma-separated, with surrounding whitespace trimmed; columns must exist. When unset or empty, it does not override identifier fields in the table schema. This is not a primary-key uniqueness constraint and does not enable upsert or delete existing rows.

​
iceberg.table.<table-name>.id-columns

Overrides the identifier column list for a single table.
  • Type: string
  • Default: Not registered; falls back to iceberg.tables.default-id-columns when read
  • Importance: N/A
  • Valid Values / Notes: An explicit empty string suppresses the global list override and preserves the table schema’s identifier fields. Other constraints are the same as for the global setting; append semantics do not change.

​
iceberg.tables.default-commit-branch

Sets the default branch for table snapshot commits.
  • Type: string
  • Default: null
  • Importance: Medium
  • Valid Values / Notes: Commits to main when unset. The branch need not exist beforehand; the underlying snapshot commit allows a new branch to be created. An existing reference with the same name must be a branch, not a tag. Per-table configuration takes precedence.

​
iceberg.table.<table-name>.commit-branch

Overrides the commit branch for a single table.
  • Type: string
  • Default: Not registered; falls back to iceberg.tables.default-commit-branch when read
  • Importance: N/A
  • Valid Values / Notes: Uses main if the effective setting remains unset. Constraints on new branches and tags are the same as for the global setting. Data written to other branches does not automatically appear in queries against main.

​
Control Channel and Commits

​
iceberg.control.topic

Specifies the control topic that carries written-file references and commit coordination events.
  • Type: string
  • Default: control-iceberg
  • Importance: Medium
  • Valid Values / Notes: Not the input data topic; the connector does not explicitly issue a topic creation request. Retention must cover events needed for unfinished commits and recovery, rather than being set solely according to the normal commit interval.

​
iceberg.control.group-id-prefix

Sets the internal control consumer group prefix.
  • Type: string
  • Default: cg-control-
  • Importance: Low
  • Valid Values / Notes: The default includes the trailing hyphen. This group differs from the Connect data consumer group, and recovery depends on retaining control consumer group offsets.

​
iceberg.control.commit.interval-ms

Sets the interval for initiating table commits, in milliseconds.
  • Type: int
  • Default: 300000
  • Importance: Medium
  • Valid Values / Notes: Use a positive value. The interval affects query visibility and file closing frequency but is not an upper bound on end-to-end latency. The configuration definition does not validate a range or the relationship between this value and the timeout.

​
iceberg.control.commit.timeout-ms

Sets the timeout for the coordinator to wait for responses from writing tasks, in milliseconds.
  • Type: int
  • Default: 30000
  • Importance: Medium
  • Valid Values / Notes: Use a positive value. On timeout, the coordinator may commit the partial set of responses already received. A timeout does not necessarily mean task failure, nor is it an upper bound on the total duration of all catalog commit operations.

​
iceberg.control.commit.threads

Sets the number of threads used by the coordinator to commit target tables in parallel.
  • Type: int
  • Default: Runtime.getRuntime().availableProcessors() * 2
  • Importance: Medium
  • Valid Values / Notes: The default expression is evaluated when the configuration definition is initialized and uses twice the number of processors visible to the worker JVM. The thread pool requires a positive value. Multiple tables are committed separately, without a cross-table transaction.

​
iceberg.kafka.*

Configures internal control Producer, Consumer, and Admin clients. This property set has no uniform default; it attempts to read worker properties, then overrides them with properties under this prefix. Individual property types are determined by the corresponding Kafka client.
  • Type: Pass-through Kafka client properties
  • Default: N/A
  • Importance: N/A
  • Valid Values / Notes: Automatic reading depends on the standard Connect startup entry point and readable worker properties containing bootstrap.servers; do not assume it works in every deployment. Explicitly set broker addresses and security properties when necessary. Worker consumer.* and producer.* properties are not automatically stripped of their prefixes. The internal Producer enforces the generated transactional ID and serializers; the internal Consumer enforces its group ID, disables automatic commits, uses read_committed and fixed deserializers, and falls back to latest for auto.offset.reset when not explicitly set. These properties do not override the task’s data consumer.

​
Advanced Identity Configuration

​
iceberg.connect.group-id

Informs the coordinator of the actual Connect data consumer group.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid Values / Notes: Falls back to connect-<connector name> when read if unset. This setting does not change the task’s consumer group. If the actual data group is changed through consumer.override.group.id, both must match; otherwise, coordinator election and commits may not complete correctly.

​
iceberg.coordinator.transactional.prefix

Sets the prefix used when generating transactional IDs.
  • Type: string
  • Default: null
  • Importance: Low
  • Valid Values / Notes: Falls back to an empty string when read. This is not a complete transactional ID; iceberg.kafka.transactional.id cannot replace the transactional ID generated by the connector.

​
Best Practices

​
Write Known Event Categories to Separate Tables

Business scenario: A single event stream contains order and payment events, and analysts need to query the two histories separately rather than have every table receive all events. Prepare analytics.orders and analytics.payments in advance, each with fields for its corresponding events. Configuration example: Override the target tables in the Quick Start configuration and add the following routing settings; retain the other catalog, converter, and Kafka settings. Use order or payment for event_type in input values.
Key considerations: Both expressions match the entire string, so each record with a valid category is written only to the corresponding table. This does not automatically map topics to table names. Records with a missing, null, or unmatched category are not written to any table, but consumer offsets may still advance; the input side must define category conventions and how to handle unmatched data. If expressions overlap, the same record is written to multiple tables, and their snapshots are not guaranteed to become visible simultaneously.

​
Automatically Create Tables for New Event Streams and Accept New Fields

Business scenario: A new audit event stream has no target table yet. Records contain a stable business-date field, and fields needed for queries will be added over time. Allow the connector to create the table from the first event and extend its schema when new fields appear in subsequent events, avoiding manual table changes before each field addition. Configuration example: In the Quick Start configuration, replace the target table with the nonexistent analytics.audit_events and add automatic creation, evolution, and date-field partitioning settings. The analytics namespace must be available, or the catalog must permit its creation.
On the input side, write JSON objects without schema envelopes to the topic specified in Quick Start. The first record contains a stable string date for partitioning, and the next adds an actor field, for example:
Key considerations: This example partitions by the original value of event_date; it does not automatically interpret the string as a timestamp or generate a business date from write time. The input side must keep the date format consistent. The first record determines the initial schema, and a subsequent non-null string actor can be added as a new field; this does not imply support for arbitrary type changes. Partition rules apply only to newly created tables. Check the actual creation result, because invalid rules may fall back to an unpartitioned table. Enabling automatic creation does not recreate or repartition existing tables.

​
Monitoring

​
What to Monitor

Monitor Kafka Connect cluster health, connector and task status, throughput, processing latency, consumer lag, offset commits, errors and retries, and worker JVM heap, GC, and thread signals. Monitor DLQ activity only when the corresponding framework error handling is enabled. Distinguish input offset advancement from target table query visibility; low lag does not directly indicate that all writes are complete.

​
Import the Grafana Dashboard

Download the Kafka Connect Cluster Grafana Dashboard. Confirm that monitoring collection is enabled and that the Prometheus-compatible data source, collected metrics, and labels match the dashboard queries, then import the JSON into Grafana and select the corresponding data source.

​
Limitations

  • The built-in write path appends records; it does not perform primary-key-based upsert, enforce uniqueness, or apply CDC row-level deletes. Null tombstones do not delete existing rows.
  • The static target table list and dynamic routing mode are mutually exclusive.
  • Snapshots for multiple tables are committed independently, without cross-table atomicity or simultaneous visibility guarantees.
  • Schema evolution does not automatically rename fields, delete columns, narrow types, or perform arbitrary breaking changes.
  • Recovery depends on retaining control events, related consumer group offsets, referenced data files, and snapshot checkpoint history. Cleaning up these resources or resetting identities or offsets may invalidate the prerequisites for exactly-once recovery.
  • Fault tolerance and DLQ configuration for the Connect Converter and SMT stages cannot be generalized to a guarantee that every bad record in the Iceberg write path can be skipped or sent to a DLQ.

​
FAQ

​
Why Is Data Still Missing from the Table After Consumer Lag Decreases?

Input offsets and file references are committed through Kafka transactions first; table snapshots are subsequently committed by the coordinator. First check whether the commit interval has elapsed and whether the target table and query branch are correct, then check for coordinator, catalog, or storage errors. The default commit interval is 5 minutes; slow responses or catalog commits can further affect visibility. Consumer offsets cannot substitute for table commit results.

​
Why Are Some Records Missing from Tables Even Though the Task Has Not Failed?

Check whether the routing field is missing or null, or fails to match the entire regular expression in static routing. Null tombstones are also not written to tables. In dynamic routing with automatic table creation disabled, writes are also skipped when the target table does not exist; this does not include invalid table identifiers, permission errors, or other catalog exceptions. These skips are not automatically sent to a DLQ, and their consumer offsets can be committed when a subsequent Kafka transaction succeeds. After correcting category data or match conditions, or creating the missing dynamic target table, a normal restart does not automatically recover skipped records whose offsets have already been committed. Determine the backfill scope and an idempotent processing strategy separately. In static routing with automatic table creation disabled, a missing target actually selected by routing puts the task in the FAILED state; it cannot be treated as a dynamic missing-table skip. Check the target table named in the exception and verify the namespace, schema, partitioning, and permissions. After creating target resources that can be loaded and are compatible with the input, keep the connector name, actual data consumer group, iceberg.connect.group-id, control topic, control consumer group and its offsets, target table and commit branch, and snapshot checkpoint identity unchanged, then restart the failed task to continue processing. First verify the persistent consumer offsets for the failed input; do not treat in-memory offset advancement as committed. After restart, check replay of the failed input, the final row set covering both failed and new input, snapshot checkpoints, and persistent offsets to confirm there are no omissions or duplicates. Retain the control events, referenced data files, and checkpoint history needed for recovery. Do not blindly switch groups or reset offsets, or treat this procedure as a guarantee of atomic multi-table recovery under arbitrary failures.

​
Why Is the Table Not Partitioned as Expected After Enabling Automatic Creation?

Partition configuration is used only when creating a missing table; existing tables are not repartitioned. For a new table, check whether the partition field’s type can be inferred, whether transforms are compatible with the field type, and whether logs contain partition rule errors. Parsing or construction exceptions may result in an unpartitioned table. Act on the actual table structure; continued writes do not mean that partition configuration succeeded.

​
Why Are There Multiple Rows for the Same Business Key After Setting Identifier Columns?

Identifier columns do not enable uniqueness constraints or merge logic; the built-in path appends every non-null record. CDC envelope normalization only organizes change records and operation metadata; it does not automatically apply updates or deletes to existing rows. If you need a current-state table, implement merging through downstream query logic or a separate processing job. Exactly-once semantics also do not deduplicate business events that are already duplicated in the input.

​
Why Do Commits Stall After Changing the Consumer Group?

Check whether the task’s actual data consumer group matches iceberg.connect.group-id. Without an override, the latter falls back to connect-<connector name>; changing only iceberg.connect.group-id does not change the actual consumer. If consumer.override.group.id is configured, ensure that both match and that the worker permits the override. Consumer groups, the control topic, and control group offsets are part of the recovery identity; arbitrary changes or resets should not be treated as transparent restarts.

​
Why Do Bad Records Still Fail the Task After Configuring a DLQ?

First distinguish whether the exception comes from the Converter or SMT, or from Iceberg record conversion, file writes, or catalog operations. Framework fault tolerance and DLQ handling cover only the stages they manage; they cannot be promised to handle this connector’s SinkTask.put write exceptions. Correct the input or target schema based on the full exception and safe topic, partition, and offset context. For storage, transaction, and catalog errors, also check the corresponding services and permissions. Avoid merely adding fault-tolerance settings and losing sight of the actual cause.

​
Why Do Queries Against main Remain Unchanged After Committing to Another Branch?

Check the effective global and per-table commit-branch values; per-table configuration takes precedence, and commits go to main only when the setting is unset. Queries must target the branch actually receiving writes. A new branch need not be created beforehand, but an existing reference with the same name cannot be a tag if it is to serve as a commit target. This setting does not automatically merge branch snapshots back into main.