org.apache.iceberg.connect.IcebergSinkConnector to sink data from kafka to databricks - large volume

[dawidch]

Hey.

We have different number of topics per let's call it "subject". The connector works great for us if there are not so many topics (partitions) in subject. We have issue when we try to sink 1500 topics, 3 partitions each. We've sharded the topics into 15 connector but it didn't help.

What happens:

the connector is starting and consuming topics for over than hour. After that we are getting the issue:

com.azure.storage.blob.models.BlobStorageException: If you are using a StorageSharedKeyCredential, and the server returned an error message that says 'Signature did not match', you can compare the string to sign with the one generated by the SDK. To log the string to sign, pass in the context key value pair 'Azure-Storage-Log-String-To-Sign': true to the appropriate method call.If you are using a SAS token, and the server returned an error message that says 'Signature did not match', you can compare the string to sign with the one generated by the SDK. To log the string to sign, pass in the context key value pair 'Azure-Storage-Log-String-To-Sign': true to the appropriate generateSas method call.Please remember to disable 'Azure-Storage-Log-String-To-Sign' before going to production as this string can potentially contain PII.Status code 403, "<?xml version="1.0" encoding="utf-8"?><Error><Code>AuthenticationFailed</Code><Message>Server failed to authenticate the request. Make sure the value of Authorization header is formed correctly including the signature.RequestId:37fc41aa-c01e-00c2-73c8-4ccf12000000Time:2026-09-25T08:35:52.7458182Z</Message><AuthenticationErrorDetail>Signature not valid in the specified time frame: Start [Fri, 25 Sep 2026 06:48:07 GMT] - Expiry [Fri, 25 Sep 2026 08:34:15 GMT] - Current [Fri, 25 Sep 2026 08:35:52 GMT]</AuthenticationErrorDetail></Error>

and the task is failing, no data is shown in databricks. Tables are created though. Here is our configuration:

{ "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector", "iceberg.catalog.cache-enabled": "false", "iceberg.catalog.credential": "${secrets:<secret-path>:<credential-key>}", "iceberg.catalog.oauth2-server-uri": "https://<databricks-workspace>/oidc/v1/token", "iceberg.catalog.rest.auth.type": "oauth2", "iceberg.catalog.token-exchange-enabled": "false", "iceberg.catalog.token-refresh-enabled": "true", "iceberg.catalog.type": "rest", "iceberg.catalog.uri": "https://<databricks-workspace>/api/2.1/unity-catalog/iceberg-rest", "iceberg.catalog.warehouse": "<warehouse>", "iceberg.catalog.scope": "all-apis", "iceberg.control.commit.interval-ms": "300000", "iceberg.control.commit.threads": "8", "iceberg.control.commit.timeout-ms": "120000", "iceberg.control.group-id-prefix": "<group-id-prefix>", "iceberg.control.topic": "<control-topic>", "iceberg.coordinator.transactional.prefix": "<transactional-prefix>", "iceberg.tables.auto-create-enabled": "true", "iceberg.tables.dynamic-enabled": "true", "iceberg.tables.evolve-schema-enabled": "true", "iceberg.tables.route-field": "icebergTable", "iceberg.tables.schema-force-optional": "true", "name": "<connector-name>", "tasks.max": "4", "topics.regex": "^(?!.*changelog)<topic-prefix>.*$", "transforms": "RouteTable,InsertTable", "transforms.InsertTable.topic.field": "icebergTable", "transforms.InsertTable.type": "org.apache.kafka.connect.transforms.InsertField$Value", "transforms.RouteTable.regex": "^<topic-prefix>(cc|cm)-(.*)$", "transforms.RouteTable.replacement": "<table-prefix>-$1.$2", "transforms.RouteTable.type": "org.apache.kafka.connect.transforms.RegexRouter", "version": "<version>" }

What can you advice?

[aayush_410]

This is almost certainly a vended-credential expiry issue on the Unity Catalog Iceberg REST catalog side, not a connector misconfiguration — and it's a known, currently-open gap in how Iceberg REST clients handle Unity Catalog credentials.

Root cause

When your connector loads an Iceberg table via the UC REST catalog, UC vends a short-lived storage credential (an ADLS SAS token) scoped to that table's path. The problem: Iceberg REST clients hold onto the same table/FileIO and the credentials it was given at load time — they don't request fresh credentials before the session expires, so once the vended token's lifetime is up, any read or write against that table fails, even though the client is still actively holding the table. This has been confirmed as a general gap across S3, ADLS, and GCS-backed UC deployments, not just Azure.

In your logs: Expiry [08:34:15] vs Current [08:35:52] — the SAS token issued when your connector loaded/created the table simply ran out before the actual blob write happened, because your connector took over an hour to get from "start consuming" to "first commit."

Why it's taking over an hour for you specifically

1500 topics × 3 partitions = 4500 partitions, split across 15 connectors → 300 partitions per connector, but each connector only has tasks.max: 4. That's ~75 partitions per task doing consumption, buffering, and (likely) table auto-creation.
iceberg.tables.auto-create-enabled: true + dynamic-enabled: true + route-field means the connector may be creating/loading hundreds of distinct tables per connector on cold start — each createTable/loadTable call is a catalog round-trip, and each one gets its own vended credential with its own expiry clock ticking from the moment it's issued.

With only 4 tasks handling all of that DDL + consumption + buffering before the 5-minute commit interval can even kick in meaningfully, you're blowing past the token's ~1-2 hour TTL before the actual write to blob storage occurs.
What to do Scale tasks.max up significantly, per connector — not just sharding into more connectors. Sharding into 15 connectors with tasks.max: 4 each only gives you 60 tasks for 4500 partitions. Push this much higher (bounded by your Connect cluster's available cores) so each task owns far fewer partitions/tables and can reach first-write inside the credential window.

Pre-create the Iceberg tables ahead of time instead of relying on auto-create-enabled at runtime for all 1500+ tables simultaneously. Run a one-time warm-up job (or DDL script) that creates every target table before you start the connector, then you can leave auto-create-enabled: true just for genuinely new tables going forward. This removes a huge chunk of the cold-start catalog overhead that's currently eating into your credential lifetime.
Shrink the time-to-first-commit — your commit.interval-ms (5 min) isn't the bottleneck; the bottleneck is everything happening before the first commit cycle can complete. Fixes 1 and 2 directly address this.
Ask your Databricks/UC admin whether the vended-credential/OAuth token TTL can be raised at the workspace or metastore level. Some REST catalog implementations expose an "OAuth/credential expiry" setting (commonly defaulting to ~1h, configurable in a 15m–12h range) — worth confirming whether your workspace admin has any control over this for UC's Iceberg REST endpoint specifically, since it would buy you headroom even without fixing the underlying refresh gap.
Track this as a known limitation, not just your config — there's an open issue on this exact behavior for Unity Catalog's Iceberg REST credential vending. Realistically, no amount of connector tuning eliminates the risk entirely for very long-running or very high-partition-count jobs until a fix (automatic credential refresh) lands upstream. Your practical mitigation today is keeping every task's load light enough that table-load-to-write always completes well inside the token's lifetime.

[dawidch]

Thanks for quick replay. Is there an open ticket for this somewhere?

[ivanvyd]

@dawidch, there is a related upstream ticket: Apache Iceberg #15852⁠, covering scheduled refresh of credentials held by ADLSFileIO. The proposed implementation is still open. I'd treat it as a lead, not yet a confirmed match for your connector version.

Databricks documents a one-hour default lifetime⁠ for the vended ADLS SAS, consistent with the timing in your log. Could you share the connector version and the Iceberg version bundled with it?

For any tuning workaround, I'd also test continuous ingestion beyond credential expiry. Getting the first commit through faster might avoid the startup failure, but it wouldn't establish that a long-running connector refreshes its storage credentials.

[dawidch]

Thanks for replay!

I've tried with ootb 11.1.0 connector with the same bundled iceberg. I've seen that github pull request, I've downloaded the code, built it and run, but I've got the same issue. It seems like the fix doesn't work for me or I did sth wrong.

[Louis_Frolio]

Hello @dawidch, I took a look at both internal and external documentation and here is what I found.

@aayush_410 Aayush and @ivanvyd Ivan have the diagnosis right, so I'll add the missing piece rather than repeat them. The 403 is an expired vended ADLS SAS credential during the data-plane write, not a broken OAuth setup. Your log shows the token expiring at 08:34:15 UTC and the failing request at 08:35:52 UTC, and Databricks documents a one-hour default lifetime for credentials returned by the Iceberg REST endpoint (returned as expires-at-ms in the loadTable response).

Here's why I suspect the rebuilt connector didn't change anything. The scheduled-refresh work in Iceberg (PR #15678 for S3, and the ADLS request in issue #15852 that Ivan linked) relies on the catalog advertising a refresh endpoint, adls.refresh-credentials-endpoint or adls.refresh-credentials-enabled, in the table's config. If the server doesn't send it, the client has nothing to refresh against and a patched ADLSFileIO behaves like the stock one. The sample response in the Databricks docs shows only expires-at-ms and the SAS token, no refresh endpoint. That's a sample, not a guarantee, so run the curl from the docs against one of your tables and look at the config block:

curl -X GET -H "Authorization: Bearer $OAUTH_TOKEN" -H "Accept: application/json" \
https://<workspace>/api/2.1/unity-catalog/iceberg-rest/v1/catalogs/<catalog>/namespaces/<schema>/tables/<table>

If there's no adls.refresh-credentials-* key, the fix isn't in your build. It's a server-side question for Databricks, and I'd raise it with your account team or a support ticket, referencing the REST spec's /credentials endpoint and issue #15852. Also confirm which Iceberg JAR the Connect worker actually loaded, since a stale artifact on the classpath would give you the same result as no patch.

While that's being sorted, the practical mitigation is Aayush's: make sure nothing holds a loaded table anywhere near an hour. You've got iceberg.catalog.cache-enabled set to false, so every fresh loadTable gets a fresh token, and the question becomes what's outliving a commit cycle. A few things to check in the logs:

• Are commit cycles completing? Look for the coordinator's commit request and commit complete messages every five minutes. If the first commit never finishes, writers created at startup keep their original token until it dies.
• How long are consumer group rebalances taking? 4,500 partitions over 4 tasks per connector on a topics.regex subscription is a lot of churn, and that delays the first commit.
• Is the coordinator hitting iceberg.control.commit.timeout-ms (120 s) committing hundreds of tables with 8 threads? If so, raise it and cut the table count per connector.

Concretely: pre-create the 1,500 tables with a one-time DDL job, raise tasks.max well above 4 within your Connect cluster's capacity, then run a two-part test. First, confirm the initial commit lands well under an hour. Second, keep the connector running past the hour mark and see whether the next write succeeds without a task restart. If part one passes and part two fails near expiry, the remaining problem is credential refresh, not startup load. Ivan's caution stands: a faster first commit avoids the startup failure but doesn't prove a long-running writer survives. I also wouldn't count on raising the credential TTL; the public docs describe the default but don't document a workspace or metastore setting for changing it.

Two housekeeping notes. Don't enable Azure's string-to-sign logging in production, since the error message itself warns it can expose sensitive data. And for a follow-up, please share the exact connector artifact and version (the "11.1.0" you mentioned, and which distribution), the bundled Iceberg version, and a redacted catalog response showing expires-at-ms.

References:

• Access Databricks tables from Iceberg clients (Azure): https://learn.microsoft.com/en-us/azure/databricks/external-access/iceberg
• Unity Catalog credential vending for external systems: https://learn.microsoft.com/en-us/azure/databricks/external-access/credential-vending
• Iceberg REST catalog OpenAPI spec (see the /credentials endpoint): https://github.com/apache/iceberg/blob/master/open-api/rest-catalog-open-api.yaml
• Iceberg issue #15852, scheduled refresh for ADLSFileIO credentials: https://github.com/apache/iceberg/issues/15852
• Iceberg PR #15678, the equivalent S3FileIO change: https://github.com/apache/iceberg/pull/15678
• Iceberg Kafka Connect sink docs: https://iceberg.apache.org/docs/latest/kafka-connect/

Regards, Louis.

[dawidch · Accepted solution]

Hello All.

Thank you all for help. Here is what worked for us:

## Original symptoms

Each connector started normally and consumed records for more than an hour. Kafka Connect reported
the connector and tasks as `RUNNING`, and source-topic lag decreased. Tables were created in
Databricks, but no rows became visible.

After the connector had been running for longer than the lifetime of its short-lived Azure Storage
SAS token, a task failed with an error similar to:

```text
com.azure.storage.blob.models.BlobStorageException: Status code 403
AuthenticationFailed: Signature not valid in the specified time frame
Start [Fri, 25 Sep 2026 06:48:07 GMT]
Expiry [Fri, 25 Sep 2026 08:34:15 GMT]
Current [Fri, 25 Sep 2026 08:35:52 GMT]
```

At first, this looked like a SAS refresh problem.

## What we tried next

We've tried custom built jar with the fix for refresh SAS token, mentioned here, in the issue but no luck. 
Then we generated a long-lived SAS token for the Azure Storage account receiving the Iceberg data files
and configured it explicitly in the connector:

```yaml
iceberg.catalog.adls.sas-token.<storage-account>.dfs.core.windows.net: <long-lived-sas-token>
```

This removed the SAS-expiration errors. The connector continued running, Kafka metrics showed that
records were being consumed, and data files were uploaded to Azure Storage. However, no rows became
visible in Databricks.

That distinction was important: the connector could consume Kafka records and upload physical
files, but those files remained uncommitted. Databricks only exposes files referenced by a committed
Iceberg snapshot.

After ruling out SAS expiration as the primary failure, we noticed this warning in the connector
logs:

```text
found no partitions assigned across all members, cannot determine leader
```

We then investigated which leader the message referred to, which Kafka consumer group Iceberg was
describing, and which group the sink tasks had actually joined. That investigation exposed the
group-ID mismatch described below.

## Configuration context

The relevant parts of our original sanitized configuration looked like this:

```json
{
  "connector.class": "org.apache.iceberg.connect.IcebergSinkConnector",
  "name": "<connector-name>",
  "tasks.max": "4",
  "topics.regex": "^(?!.*changelog)<topic-prefix>.*$",

  "iceberg.catalog.type": "rest",
  "iceberg.catalog.uri": "https://<databricks-workspace>/api/2.1/unity-catalog/iceberg-rest",
  "iceberg.catalog.rest.auth.type": "oauth2",
  "iceberg.catalog.oauth2-server-uri": "https://<databricks-workspace>/oidc/v1/token",
  "iceberg.catalog.credential": "${secrets:<secret-path>:<credential-key>}",
  "iceberg.catalog.warehouse": "<warehouse>",
  "iceberg.catalog.scope": "all-apis",
  "iceberg.catalog.token-exchange-enabled": "false",
  "iceberg.catalog.token-refresh-enabled": "true",
  "iceberg.catalog.cache-enabled": "false",

  "iceberg.control.topic": "<control-topic>",
  "iceberg.control.group-id-prefix": "<group-id-prefix>",
  "iceberg.coordinator.transactional.prefix": "<transactional-prefix>",
  "iceberg.control.commit.interval-ms": "300000",
  "iceberg.control.commit.timeout-ms": "120000",
  "iceberg.control.commit.threads": "8",

  "iceberg.tables.dynamic-enabled": "true",
  "iceberg.tables.auto-create-enabled": "true",
  "iceberg.tables.evolve-schema-enabled": "true",
  "iceberg.tables.schema-force-optional": "true",
  "iceberg.tables.route-field": "icebergTable",

  "transforms": "RouteTable,InsertTable",
  "transforms.RouteTable.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.RouteTable.regex": "^<topic-prefix>(cc|cm)-(.*)$",
  "transforms.RouteTable.replacement": "<table-prefix>-$1.$2",
  "transforms.InsertTable.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.InsertTable.topic.field": "icebergTable"
}
```

One crucial detail was missing from the original sanitized configuration: the effective connector
configuration also used a custom sink consumer group through
`consumer.override.group.id`, while `iceberg.connect.group-id` was unset.

## What the leader message means

The leader mentioned in the warning is the Iceberg sink's internal commit coordinator. It is not a
Kafka broker partition leader and not the Kafka Connect worker leader.

One sink task must become the Iceberg commit coordinator. The coordinator:

1. Sends a `StartCommit` event through the Iceberg control topic.
2. Makes workers finalize their current data files.
3. Collects `DataWritten` and `DataComplete` responses from all workers.
4. Commits those files as an Iceberg snapshot.
5. Publishes a `CommitComplete` event.

Without this coordinator, sink tasks can continue accepting records and writing or buffering data,
but no data files are added to a visible Iceberg snapshot.

## Root cause

Our effective configuration was equivalent to:

```yaml
consumer.override.group.id: my-iceberg-sink-group
```

but it did not set:

```yaml
iceberg.connect.group-id: my-iceberg-sink-group
```

The source consumers therefore joined `my-iceberg-sink-group`. The Iceberg coordinator-election
code uses `iceberg.connect.group-id` when it calls Kafka's `DescribeConsumerGroups` API to inspect
the task assignments. When that property is unset, Iceberg defaults to:

```text
connect-<connector-name>
```

Iceberg was consequently comparing two different groups:

```text
Sink tasks consumed with:       my-iceberg-sink-group
Iceberg inspected for election: connect-<connector-name>
```

Kafka correctly reported no active partition assignments for the second group. The actual sink
consumer group had active members and assigned source partitions, but Iceberg never inspected it.
No task could determine that it owned the first assigned partition, so no commit coordinator was
elected.

Retrying leader election did not solve this condition because every retry queried the same wrong
consumer group.

## Why topic count and sharding looked responsible

The large number of topics made the problem much more visible, but it was not the root cause.

With approximately 4,500 partitions, initial consumption and file writing continued for a long
time. This made it look as though the connector was successfully processing a large backlog. In
reality, Kafka Connect's record counters only proved that records had been delivered to the sink
tasks; they did not prove that an Iceberg snapshot had been committed.

Splitting the topics into 15 connectors did not help because the sharded connectors retained the
same group-ID mismatch. Each connector could consume its assigned topics, but none could identify
its Iceberg commit coordinator.

Subjects with fewer topics either used a configuration without the custom consumer-group override
or had matching group IDs. The apparent relationship with topic count was therefore a configuration
correlation rather than an Iceberg scalability limit.

## Why the expired SAS token was not the root cause

The short-lived SAS token worked throughout its valid lifetime. During that time, workers could
upload data files to Azure Storage.

Because no Iceberg coordinator was running, no commit cycle finalized those files and added them to
an Iceberg snapshot. The write activity continued without completing the expected commit lifecycle
and eventually outlived the short-lived credential.

Once the SAS token expired, Azure Storage rejected additional writes. The connector then failed
with the 403 authentication error, but it had already been unable to publish data because no
coordinator was committing snapshots.

The actual sequence was:

1. Iceberg inspected the wrong consumer group and could not elect a coordinator.
2. Workers consumed records and uploaded files while the SAS token remained valid.
3. No Iceberg snapshot commit completed.
4. The long-running write activity outlived the SAS token.
5. The expired token prevented further storage writes and caused the task to fail.

Replacing, extending, or refreshing the SAS token could temporarily allow file writes to continue,
but it could not make the uncommitted files visible in the table. The group-ID mismatch had to be
fixed first.

Table creation and schema evolution were also misleading signals. Those catalog operations can
complete independently from the coordinated data-file snapshot commit, which explains why empty
tables appeared in Databricks even though no rows were visible.

## Fix

We explicitly configured Iceberg to inspect the same group used by the sink consumers:

```yaml
consumer.override.group.id: &sink_group_id my-iceberg-sink-group
iceberg.connect.group-id: *sink_group_id
```

Our complete group-related naming convention is:

```yaml
consumer.override.group.id: &sink_group_id my-iceberg-sink-group
iceberg.connect.group-id: *sink_group_id
iceberg.control.topic: *sink_group_id
iceberg.control.group-id-prefix: my-iceberg-sink-group-
iceberg.coordinator.transactional.prefix: my-iceberg-sink-group-
```

The YAML anchor is optional. We use it to prevent the two critical group-ID properties from
drifting apart.

After redeploying with matching group IDs, Iceberg immediately saw the real task assignments,
elected a commit coordinator, and began committing data successfully.

## How to verify the mismatch

In Kafka UI or with `kafka-consumer-groups.sh`, compare:

1. The group configured by `consumer.override.group.id`.
2. The effective `iceberg.connect.group-id`, or `connect-<connector-name>` when it is unset.

For example:

```bash
kafka-consumer-groups.sh \
  --bootstrap-server <bootstrap-server> \
  --command-config <client.properties> \
  --describe \
  --group my-iceberg-sink-group \
  --members \
  --verbose

kafka-consumer-groups.sh \
  --bootstrap-server <bootstrap-server> \
  --command-config <client.properties> \
  --describe \
  --group connect-<connector-name> \
  --members \
  --verbose
```

In our case, the first group showed active members and source-topic assignments. The second group
was empty or had no active assignments, exactly matching the Iceberg warning.

## Other misleading signals

- Kafka Connect's `RUNNING` state means that the task thread has not failed. It does not guarantee
  that the Iceberg commit protocol is completing.
- Decreasing source lag, `sink-record-read-total`, and successful Kafka Connect offset commits do
  not prove that an Iceberg snapshot was committed.
- `iceberg.control.group-id-prefix` configures control-channel consumers. It does not replace
  `iceberg.connect.group-id` for source-task assignment discovery.
- `RoundRobinAssignor` was not the cause. It only determines how source partitions are distributed
  among the members of the actual sink consumer group.
- Increasing storage performance, catalog retry counts, or the number of connector tasks cannot
  correct a consumer-group identity mismatch.

The definitive success signals were the Iceberg coordinator's completed-commit messages and new
snapshots becoming visible in the catalog.