Happened in the search update pipeline when upgrading to a new schema:
java.lang.IllegalArgumentException: Unknown field name 'page_type' for mapping to a position. at org.apache.flink.types.Row.getField(Row.java:292) at org.wikimedia.eventutilities.flink.EventRowSerializer.copy(EventRowSerializer.java:153) at org.wikimedia.eventutilities.flink.EventRowSerializer.copy(EventRowSerializer.java:90) at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.copyField(PojoSerializer.java:326) at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.copy(PojoSerializer.java:262) at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.copyField(PojoSerializer.java:326) at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.copy(PojoSerializer.java:262) at org.apache.flink.api.java.typeutils.runtime.EitherSerializer.copy(EitherSerializer.java:96) at org.apache.flink.api.java.typeutils.runtime.EitherSerializer.copy(EitherSerializer.java:38) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:74) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:50) at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29) at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:53) at org.apache.flink.streaming.api.operators.async.queue.StreamRecordQueueEntry.emitResult(StreamRecordQueueEntry.java:64) at org.apache.flink.streaming.api.operators.async.queue.OrderedStreamElementQueue.emitCompletedElement(OrderedStreamElementQueue.java:71) at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator.outputCompletedElement(AsyncWaitOperator.java:393) at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.processResults(AsyncWaitOperator.java:632) at org.apache.flink.streaming.api.operators.async.AsyncWaitOperator$ResultHandler.lambda$processInMailbox$0(AsyncWaitOperator.java:613) [...]
EventRowSerializer does use COMPATIBLE_AFTER_MIGRATION but this strategy is only taken into account with rocksdb/ForStStateBackend keyed state backends. I suspect that the reason is that with these backends the state is not guaranteed to be fully rewritten on every checkpoints/savepoints and that an explicit migration step is mandatory to avoid keeping very old versions in the state.
For operator states or the HashMapState backend I suspect that the migration will actually happen on the next checkpoint/savepoint.
A migration might also occur implicitly when pushing the record to downstream operators (via CopyingChainingOutput.pushToOperator) but in this case flink appears to use Serializer::copy and this is actually where EventRowSerializer failed, it did not expect to have to copy an instance obtained from an old version of the serializer.
In short my understanding is that COMPATIBLE_AFTER_MIGRATION behaves similarly to COMPATIBLE_AS_IS for:
- operator state
- HashMapState keyed states
EventRowSerializer.copy should be adapted to support copying from an older version.
This also means that we cannot use object-reuse (no migration will happen at all) unless the pipeline is aware of this behavior.