[BUG] Exception thrown by iceberg when using table topic when event schema changed

[dumbNickname]

AutoMQ Version

v1.5.5

Operating System

Docker on MacOS

Installation Method

docker

Hardware Configuration

No response

Other Relevant Software

https://raw.githubusercontent.com/AutoMQ/automq/refs/tags/1.5.5/docker/table_topic/docker-compose.yml

with akhq added

What Went Wrong?

When experimenting with producing events that seemed to follow compatible schema evolution (at least according to my understanding) the event is accepted by automq and schema registry but an error is thrown when iceberg tries to process it. Later no other events get processed and error is thrown in the background all the time as it seems to be triggered from an EventLoop

automq           | java.lang.ArrayIndexOutOfBoundsException: Index 9 out of bounds for length 9
automq           |      at org.apache.iceberg.data.GenericRecord.get(GenericRecord.java:114)
automq           |      at org.apache.iceberg.data.parquet.GenericParquetWriter$RecordWriter.get(GenericParquetWriter.java:48)
automq           |      at org.apache.iceberg.data.parquet.GenericParquetWriter$RecordWriter.get(GenericParquetWriter.java:41)
automq           |      at org.apache.iceberg.parquet.ParquetValueWriters$StructWriter.write(ParquetValueWriters.java:580)
automq           |      at org.apache.iceberg.parquet.ParquetWriter.add(ParquetWriter.java:135)
automq           |      at org.apache.iceberg.io.DataWriter.write(DataWriter.java:71)
automq           |      at org.apache.iceberg.io.BaseTaskWriter$RollingFileWriter.write(BaseTaskWriter.java:392)
automq           |      at org.apache.iceberg.io.BaseTaskWriter$RollingFileWriter.write(BaseTaskWriter.java:375)
automq           |      at org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.write(BaseTaskWriter.java:307)
automq           |      at org.apache.iceberg.io.PartitionedFanoutWriter.write(PartitionedFanoutWriter.java:63)
automq           |      at kafka.automq.table.worker.IcebergWriter.write0(IcebergWriter.java:149)
automq           |      at kafka.automq.table.worker.IcebergWriter.write(IcebergWriter.java:96)
automq           |      at kafka.automq.table.worker.PartitionWriteTask.handleReadResult(PartitionWriteTask.java:121)
automq           |      at kafka.automq.table.worker.PartitionWriteTask.lambda$run0$0(PartitionWriteTask.java:70)
automq           |      at kafka.automq.table.worker.EventLoops$EventLoopRef.lambda$execute$0(EventLoops.java:142)
automq           |      at kafka.automq.table.worker.EventLoops$EventLoopRef.lambda$execute$1(EventLoops.java:154)
automq           |      at com.automq.stream.utils.threads.EventLoop.run(EventLoop.java:69)
automq           | [2025-11-13 22:05:10,462] ERROR [HANDLE_READ_RESULT_FAIL],dedw000000000128_events_page_view-0,2-3 (kafka.automq.table.worker.PartitionWriteTask)
automq           | java.io.IOException: java.lang.ArrayIndexOutOfBoundsException: Index 9 out of bounds for length 9
automq           |      at kafka.automq.table.worker.IcebergWriter.write(IcebergWriter.java:115)
automq           |      at kafka.automq.table.worker.PartitionWriteTask.handleReadResult(PartitionWriteTask.java:121)
automq           |      at kafka.automq.table.worker.PartitionWriteTask.lambda$run0$0(PartitionWriteTask.java:70)
automq           |      at kafka.automq.table.worker.EventLoops$EventLoopRef.lambda$execute$0(EventLoops.java:142)
automq           |      at kafka.automq.table.worker.EventLoops$EventLoopRef.lambda$execute$1(EventLoops.java:154)
automq           |      at com.automq.stream.utils.threads.EventLoop.run(EventLoop.java:69)
automq           | Caused by: java.lang.ArrayIndexOutOfBoundsException: Index 9 out of bounds for length 9
automq           |      at org.apache.iceberg.data.GenericRecord.get(GenericRecord.java:114)
automq           |      at org.apache.iceberg.data.parquet.GenericParquetWriter$RecordWriter.get(GenericParquetWriter.java:48)
automq           |      at org.apache.iceberg.data.parquet.GenericParquetWriter$RecordWriter.get(GenericParquetWriter.java:41)
automq           |      at org.apache.iceberg.parquet.ParquetValueWriters$StructWriter.write(ParquetValueWriters.java:580)
automq           |      at org.apache.iceberg.parquet.ParquetWriter.add(ParquetWriter.java:135)
automq           |      at org.apache.iceberg.io.DataWriter.write(DataWriter.java:71)
automq           |      at org.apache.iceberg.io.BaseTaskWriter$RollingFileWriter.write(BaseTaskWriter.java:392)
automq           |      at org.apache.iceberg.io.BaseTaskWriter$RollingFileWriter.write(BaseTaskWriter.java:375)
automq           |      at org.apache.iceberg.io.BaseTaskWriter$BaseRollingWriter.write(BaseTaskWriter.java:307)
automq           |      at org.apache.iceberg.io.PartitionedFanoutWriter.write(PartitionedFanoutWriter.java:63)
automq           |      at kafka.automq.table.worker.IcebergWriter.write0(IcebergWriter.java:149)
automq           |      at kafka.automq.table.worker.IcebergWriter.write(IcebergWriter.java:96)
automq           |      ... 5 more

What Should Have Happened Instead?

To be honest I am not sure. I am new to iceberg and usage of schemas in message queues. I was just researching the toy example to make sure what is needed in the setup and to understand it better. My expectation would be that either event gets rejected early or if already present in the topic (after validation in schema registry) then processed properly or at least not cause automq (or more precisely iceberg) to stop next events processing

Steps to Reproduce

1. Run the "toy" example from https://raw.githubusercontent.com/AutoMQ/automq/refs/tags/1.5.5/docker/table_topic/docker-compose.yml
2. Open the example notebook
3. In my case I simplified the topic config, probably not relevant but sharing for completeness

topic_config = {
    'automq.table.topic.enable': 'true',
    'automq.table.topic.commit.interval.ms': '10000', # 10s
    'automq.table.topic.schema.type': 'schema',
    'automq.table.topic.upsert.enable': 'false', 
    'automq.table.topic.partition.by': '[truncate(event.type, 100), hour(collect_timestamp)]'
}
create_topic(TOPIC_NAME, config=topic_config)

4. Produce any base event - I named it "original"

original_events = json.loads(json.dumps(json_events))
original_events[0]["event"]["id"] = "original"

# I slightly simplified the produce method as I do not need a key serializer and just moved the avro serlilizer there. I hope this example will be still easy to follow
produce_events(
    producer=producer,
    topic_name=TOPIC_NAME,
    events_data=original_events,
    schema_str=schema_str
)

5. Insert compatible event with one new field and a matching schema (no problems here)

compatible_event = json.loads(json.dumps(json_events))[0]
compatible_event["event"]["id"] = "compatible_event_new_field_new_schema"
compatible_event["this_field_is_new"] = "new kid on the block"

next_schema_1 = json.loads(json.dumps(avro_schema))
next_schema_1['fields'].extend([
    {
        "name": "this_field_is_new", 
        "type": ["null", "string"], # optional
        "default": None, 
        "doc": "Testing things"
    }
])


produce_events(
    producer=producer,
    topic_name=TOPIC_NAME,
    events_data=[compatible_event],
    schema_str=json.dumps(next_schema_1, indent=2)
)

6. Add event with another new field, but just without the new field that was added before

compatible_event = json.loads(json.dumps(json_events))[0]
compatible_event["event"]["id"] = "maybe_compatible_event_another_new_field_without_first_added_field"
compatible_event["another_field_is_new"] = "another kid on the block"

next_schema_2 = json.loads(json.dumps(avro_schema))
next_schema_2['fields'].extend([
    {
        "name": "another_field_is_new", 
        "type": "string", # obligatory
        "default": None, 
        "doc": "Testing things"
    }
])


produce_events(
    producer=producer,
    topic_name=TOPIC_NAME,
    events_data=[compatible_event],
    schema_str=json.dumps(next_schema_2, indent=2)
)

7. Event is properly added to the topic, new schema is created so on the message queue part all looks fine. Unfortunately in the logs exception starts to be thrown and next events being produced will be ignored. By ignored I mean I do not see those later when getting iceberg data via duckdb.

Additional Information

I am not sure if I provided all the needed information - please let me know if sth is missing. Would be great to get a response if this is a bug and some hints on what should be the expected behavior. Then maybe I can try to look at the code and if I manage to find myself I can try to help implement some fix.

Thanks in advance for your help!

[Gezi-lzq]

@dumbNickname , Thanks a ton for reporting this issue.

From my understanding, the scenario you're experimenting with is an optional field 1 addition and an optional field 2 replacing an existing field 1, right? I'll try to reproduce this scenario to confirm the problem.

I also noticed you're currently using version 1.5.5. We generally recommend testing with version 1.6.0 now. We also provide a much more convenient example/playground here: https://github.com/AutoMQ/automq-labs/tree/main/table-topic-solutions/playground

[dumbNickname]

Thanks a lot for such a quick response.

My scenario is what you described but with slightly different back story. I thought that it should be fine to add field_a to the baseline schema (creates schema v2), next add field_b on top of the baseline schema again (creates schema v3) and then in future after alignment is done to have a new schema where field_a and field_b are included (would be schema v4). The schemas in brackets is what kafka schema registry seems to follow - so looked valid to me.

Just in case this scenario sounds odd, will try to explin my reasoning here - we have multiple apps sending events to kafka (hopefully automq in future - btw thanks for this product), which is schema less currently. Each app is developed by another team, with their own velocity and prioritization of tasks, so it might happen that some apps need event extensions independently from each other. Lets say the native app needs to add a field_a and the web team needs to add field_b. When both get released and teams sync about extensions they did the schema will be aligned to include both, but would be nice that before that happens each team can work on their use case and not be blocked (or need information from another team) that they work on own schema extension. As you can imagine information flow between teams is not alwyas that fluent. In perfect world all could be agreed on upfront and schema could be extended with both fields in both teams but we know how it will be in practice :P Plus during development sb might suddenly rename the field or add one more so the teams would need to stay in constant sync - what is some option but less dependencies feels better. Sorry for this too long story.

If this scenario is nonsense and you do not see it as a valid schema evolution, would be great to at least avoid iceberg crashes.

I just tried my example against version 1.6.0 and the error I get now is different (but it looks like you improved the error handling as this one at least says sth I can understand without knowing much about internals :D )

[2025-11-14 08:02:40,949] ERROR [HANDLE_READ_RESULT_FAIL],128_events_page_view-0,0-3 (kafka.automq.table.worker.PartitionWriteTask)
automq           | java.io.IOException: kafka.automq.table.process.exception.RecordProcessorException: Data processing failed for record: topic=128_events_page_view, partition=0, offset=2, timestamp=1763107315998 - [UNKNOW_ERROR] Unexpected error processing record: topic=128_events_page_view, key=null, offset=2, timestamp=1763107315998: Invalid default for field another_field_is_new: null not a "string" (caused by: AvroTypeException - Invalid default for field another_field_is_new: null not a "string")
automq           |      at kafka.automq.table.worker.IcebergWriter.write(IcebergWriter.java:134)
automq           |      at kafka.automq.table.worker.PartitionWriteTask.handleReadResult(PartitionWriteTask.java:121)
automq           |      at kafka.automq.table.worker.PartitionWriteTask.lambda$run0$0(PartitionWriteTask.java:70)
automq           |      at kafka.automq.table.worker.EventLoops$EventLoopRef.lambda$execute$0(EventLoops.java:160)
automq           |      at kafka.automq.table.worker.EventLoops$EventLoopRef.lambda$execute$1(EventLoops.java:172)
automq           |      at com.automq.stream.utils.threads.EventLoop.run(EventLoop.java:69)
automq           | Caused by: kafka.automq.table.process.exception.RecordProcessorException: Data processing failed for record: topic=128_events_page_view, partition=0, offset=2, timestamp=1763107315998 - [UNKNOW_ERROR] Unexpected error processing record: topic=128_events_page_view, key=null, offset=2, timestamp=1763107315998: Invalid default for field another_field_is_new: null not a "string" (caused by: AvroTypeException - Invalid default for field another_field_is_new: null not a "string")
automq           |      at kafka.automq.table.worker.IcebergWriter.write0(IcebergWriter.java:158)
automq           |      at kafka.automq.table.worker.IcebergWriter.write(IcebergWriter.java:116)
automq           |      ... 5 more
automq           | Caused by: org.apache.avro.AvroTypeException: Invalid default for field another_field_is_new: null not a "string"
automq           |      at org.apache.avro.Schema.validateDefault(Schema.java:1635)
automq           |      at org.apache.avro.Schema.access$500(Schema.java:94)
automq           |      at org.apache.avro.Schema$Field.<init>(Schema.java:561)
automq           |      at org.apache.avro.Schema$Field.<init>(Schema.java:572)
automq           |      at kafka.automq.table.process.RecordAssembler.buildFinalAssemblerSchema(RecordAssembler.java:148)
automq           |      at kafka.automq.table.process.RecordAssembler.getOrCreateAssemblerSchema(RecordAssembler.java:139)
automq           |      at kafka.automq.table.process.RecordAssembler.assemble(RecordAssembler.java:112)
automq           |      at kafka.automq.table.process.DefaultRecordProcessor.process(DefaultRecordProcessor.java:119)
automq           |      at kafka.automq.table.worker.IcebergWriter.write0(IcebergWriter.java:145)
automq           |      ... 6 more
automq           | [2025-11-14 08:02:40,949] INFO [SYNC_TASK],[COMMIT-128_events_page_view-12],size=2427.0,decompressedRatio=1.0,parallel=1,partitions=[0-0-3],cost=9ms (kafka.automq.table.worker.TopicPartitionsWorker)

Thank you very much for the link to your example/playground setup. Will try to check it today!

[Gezi-lzq]

Hey, thanks for the great background context!

That's a perfectly valid scenario and definitely one we want to support. I'll continue to follow the steps to reproduce this and work on getting it fixed.

I'll post all future updates and the eventual fix right here in this issue.

[dumbNickname]

Please let me know if I can anyhow help / provide more info etc. Meanwhile I will keep exploring.

[Gezi-lzq]

@dumbNickname I've set up a simple experimental scenario based on the process you described. Could you please check if it meets the expectations?
AutoMQ/automq-labs#100 (comment)

[dumbNickname]

Please excuse slow reponses but I try to explore your repo and understand the code as I am pretty sure I will either take some inspirations from your playgroud repo or try to post some examples myself if any come in future. I thought I will try to do my scenario today but you were faster anyway :D.

Back to your example - I am not sure if I understand why tables are created by hand (but I assume you want to enforce the schema so that later automq table topic already needs to match it instead of creating own tables). EDIT: After reading not in rush. I saw you just inspect calls to create table - so I think I follow now what happens there :P

Your examples with v2,v3,v4 look valid but as I commented on the PR there the one I wanted is v5, so the one that is also not listed on the final select statement as pure data - again not sure if missing because it is broken or just because this is still WIP. EDIT: I saw it got updated - v4 looks like what I intended to have