flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Si Chen (Jira)" <j...@apache.org>
Subject [jira] [Created] (FLINK-22608) Flink Kryo deserialize read wrong bytes
Date Sat, 08 May 2021 08:59:00 GMT
Si Chen created FLINK-22608:

             Summary: Flink Kryo deserialize read wrong bytes
                 Key: FLINK-22608
                 URL: https://issues.apache.org/jira/browse/FLINK-22608
             Project: Flink
          Issue Type: Bug
          Components: Runtime / Checkpointing, Runtime / State Backends
    Affects Versions: 1.12.0
            Reporter: Si Chen

In flink program, I use ValueState to save my state. The state stores pojo. But my pojo used
kryo serializer. As the program run some time, I add a field in pojo. Then recovery the program
with checkpoint. I found the value of the field incorrect. Then I read the source code I


private void readStateHandleStateData(
   FSDataInputStream fsDataInputStream,
   DataInputViewStreamWrapper inView,
   KeyGroupRangeOffsets keyGroupOffsets,
   Map<Integer, StateMetaInfoSnapshot> kvStatesById,
   int numStates,
   int readVersion,
   boolean isCompressed) throws IOException {

   final StreamCompressionDecorator streamCompressionDecorator = isCompressed ?
      SnappyStreamCompressionDecorator.INSTANCE : UncompressedStreamCompressionDecorator.INSTANCE;

   for (Tuple2<Integer, Long> groupOffset : keyGroupOffsets) {
      int keyGroupIndex = groupOffset.f0;
      long offset = groupOffset.f1;

      // Check that restored key groups all belong to the backend.
      Preconditions.checkState(keyGroupRange.contains(keyGroupIndex), "The key group must
belong to the backend.");


      int writtenKeyGroupIndex = inView.readInt();
      Preconditions.checkState(writtenKeyGroupIndex == keyGroupIndex,
         "Unexpected key-group in restore.");

      try (InputStream kgCompressionInStream =
             streamCompressionDecorator.decorateWithCompression(fsDataInputStream)) {

my state keyGroupIndex is 81, and keyGroupOffset is 3572. And the next keyGroupOffset is 3611.
So my state offset rang is 3572 to 3611. But when I add new field in pojo. Kryo will read
more bytes in the next keyGroup.


This message was sent by Atlassian Jira

View raw message