Flink: Dynamic Sink — ClassCastException in HashKeyGenerator partition keying when records for one table alternate between writer schema variants
- 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
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