[
https://issues.apache.org/jira/browse/FLINK-38023?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-38023:
-----------------------------------
Labels: pull-request-available (was: )
> State Restoration fails when using GenericRecord with new Avro schema
> ---------------------------------------------------------------------
>
> Key: FLINK-38023
> URL: https://issues.apache.org/jira/browse/FLINK-38023
> Project: Flink
> Issue Type: Bug
> Reporter: Keshav Kansal
> Priority: Major
> Labels: pull-request-available
>
> Hi Team,
> I have a Flink job which is storing a
> [GenericRecord|https://github.com/apache/avro/blob/main/lang/java/avro/src/main/java/org/apache/avro/generic/GenericRecord.java]
> in the ValueState and we use
> [GenericRecordAvroTypeInfo|https://github.com/apache/flink/blob/master/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/GenericRecordAvroTypeInfo.java]
> I am observing an issue while upgrading the avro schema.
> I reproduced this locally -
> For example - Initially the Schema set is
> {code:java}
> {
> "type" : "record",
> "name" : "SimpleEvent",
> "fields" : [ {
> "name" : "count",
> "type" : "int"
> } ]
> }
> {code}
> A savepoint is taken with some entities in the state.
> The app is restored from the savepoint from the previous state and the below
> new schema(Note - This is backward comptabile)
> {code:java}
> {
> "type" : "record",
> "name" : "SimpleEvent",
> "fields" : [ {
> "name" : "count",
> "type" : "int"
> }, {
> "name" : "strTest",
> "type" : [ "null", "string" ],
> "default" : null
> } ]
> }
> {code}
> While restoring the data from savepoint we observe the following exception
> {code:java}
> Caused by: org.apache.flink.util.StateMigrationException: Error while trying
> to migrate RocksDB state.
> at
> org.apache.flink.contrib.streaming.state.AbstractRocksDBState.migrateSerializedValue(AbstractRocksDBState.java:194)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.migrateStateValues(RocksDBKeyedStateBackend.java:848)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.updateRestoredStateMetaInfo(RocksDBKeyedStateBackend.java:772)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.tryRegisterKvStateInformation(RocksDBKeyedStateBackend.java:684)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.createOrUpdateInternalState(RocksDBKeyedStateBackend.java:904)
> at
> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend.createOrUpdateInternalState(RocksDBKeyedStateBackend.java:891)
> at
> org.apache.flink.runtime.state.KeyedStateFactory.createOrUpdateInternalState(KeyedStateFactory.java:47)
> at
> org.apache.flink.runtime.state.ttl.TtlStateFactory.createStateAndWrapWithTtlIfEnabled(TtlStateFactory.java:70)
> at
> org.apache.flink.runtime.state.AbstractKeyedStateBackend.getOrCreateKeyedState(AbstractKeyedStateBackend.java:361)
> at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.getOrCreateKeyedState(StreamOperatorStateHandler.java:361)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.getOrCreateKeyedState(AbstractStreamOperator.java:517)
> at
> org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.open(WindowOperator.java:238)
> at
> org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:107)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreStateAndGates(StreamTask.java:858)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$restoreInternal$5(StreamTask.java:812)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:812)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:771)
> at
> org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970)
> at
> org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:939)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
> at java.base/java.lang.Thread.run(Thread.java:840)
> Caused by: java.lang.ArrayIndexOutOfBoundsException: Index 1 out of bounds
> for length 1
> at org.apache.avro.generic.GenericData$Record.get(GenericData.java:289)
> at org.apache.avro.generic.GenericData.getField(GenericData.java:860)
> at org.apache.avro.generic.GenericData.getField(GenericData.java:879)
> at
> org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:243)
> at
> org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:234)
> at
> org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:145)
> at
> org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:95)
> at
> org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:82)
> at
> org.apache.flink.formats.avro.typeutils.AvroSerializer.serialize(AvroSerializer.java:188)
> at
> org.apache.flink.contrib.streaming.state.AbstractRocksDBState.migrateSerializedValue(AbstractRocksDBState.java:192)
> ... 22 more
> {code}
> Tested with Flink version 1.20.1
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
