apache / apache/iceberg

Flink: Dynamic Sink — ClassCastException in HashKeyGenerator partition keying when records for one table alternate between writer schema variants

Open
#17,651 0 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Java
Stars
9.2k
Forks
3.5k
Avg merge
2d 11h
Merged PRs (30d)
132

Description

### Apache Iceberg version

1.10.1

### Query engine

Flink

### Please describe the bug 🐞

We use the Dynamic Sink to ingest Confluent-Avro Kafka topics where multiple writer schema versions are live simultaneously (normal schema-registry evolution: some producers on version N, others on N+1). Per writer version we derive an Iceberg Schema (independently fresh-assigned field IDs, so the same field name has different IDs across variants) plus a PartitionSpec built against that variant via PartitionSpec.builderFor(variantSchema).identity("event_date"), and emit DynamicRecords with DistributionMode.HASH.

When records for the same table alternate between two schema variants, HashKeyGenerator's partition-key selector reads the wrong field position for the partition source — the accessor resolves a field ID from one variant against row data laid out per the other — producing:

Caused by: java.lang.ClassCastException: Cannot cast java.lang.String to java.lang.Integer
at java.base/java.lang.Class.cast(Unknown Source)
at org.apache.iceberg.StructTransform.get(StructTransform.java:89)
at org.apache.iceberg.PartitionKey.get(PartitionKey.java:30)
at org.apache.iceberg.PartitionSpec.partitionToPath(PartitionSpec.java:225)
at org.apache.iceberg.flink.sink.PartitionKeySelector.getKey(PartitionKeySelector.java:64)
at org.apache.iceberg.flink.sink.dynamic.HashKeyGenerator.lambda$partitionKeySelector$...(HashKeyGenerator.java:218)
at org.apache.iceberg.flink.sink.dynamic.HashKeyGenerator.generateKey(HashKeyGenerator.java:104)
at org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.emit(DynamicRecordProcessor.java:162)

(event_date is a DATE — internally Integer — and the accessor lands on a String column from the other variant's layout.)

Reproduction conditions

- Table partitioned by identity on a date column that is not at the same field ID/position across variants (fields differ before it in one variant).
- DynamicRecords alternate between variant A and variant B (each with its own Schema + PartitionSpec instance, IDs assigned independently).
- DistributionMode.HASH. With DistributionMode.NONE the crash does not occur (no partition-key extraction on the record path) — this is our current workaround.

Happy to provide a full runnable reproducer if useful.

Expected behavior

Keying should be performed against the record's own schema/spec (or after adaptation to the table schema), so mixed writer-schema versions — the normal state of a schema-registry-fed stream — hash correctly.

### Willingness to contribute

- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time

Contributor guide

Open the contributing guide

Research direction

Start with flink/sink/dynamic/HashKeyGenerator.java and follow its partitionKeySelector call into PartitionKeySelector.java, DynamicRecordProcessor.java, and the StructTransform/PartitionKey stack shown in the trace. Reproduce alternating records with independently assigned schema and partition-spec field IDs under DistributionMode.HASH, then verify that key extraction uses the record's schema/spec or an equivalent table-schema adaptation without the ClassCastException; DistributionMode.NONE is the current workaround.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
data-engineering, distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.