TaggedOperatorSubtaskState is missing when creating a new savepoint using state processor api

[郭斌]

Hi, 各位好：
   我在使用state processor api创建新的包含kafka相关state的savepoint用来修改max parallelism时，创建成功后使用此savepoint来重启任务，发现抛出如下异常：
  {code}
  java.lang.Exception: Exception while creating StreamOperatorStateContext.
    at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:204)
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:247)
    at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:290)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:473)
    at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:469)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.util.FlinkException: Could not restore operator state backend for StreamSource_e8ea6e352a1a627513ffbd4573fa1628_(1/1) from any of the 1 provided restore options.
    at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
    at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.operatorStateBackend(StreamTaskStateInitializerImpl.java:265)
    at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:152)
    ... 9 more
Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore operator state backend
    at org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder.build(DefaultOperatorStateBackendBuilder.java:86)
    at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createOperatorStateBackend(RocksDBStateBackend.java:552)
    at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$operatorStateBackend$0(StreamTaskStateInitializerImpl.java:256)
    at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
    at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
    ... 11 more
Caused by: java.lang.IllegalStateException: Missing value for the key 'org.apache.flink.state.api.output.TaggedOperatorSubtaskState'
    at org.apache.flink.util.LinkedOptionalMap.unwrapOptionals(LinkedOptionalMap.java:190)
    at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot.restoreSerializer(KryoSerializerSnapshot.java:86)
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
    at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
    at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
    at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225)
    at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83)
    at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:204)
    at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189)
    at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164)
    at org.apache.flink.runtime.state.RegisteredOperatorStateBackendMetaInfo.getPartitionStateSerializer(RegisteredOperatorStateBackendMetaInfo.java:113)
    at org.apache.flink.runtime.state.OperatorStateRestoreOperation.restore(OperatorStateRestoreOperation.java:94)
    at org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder.build(DefaultOperatorStateBackendBuilder.java:83)
    ... 15 more
  {code}
我的使用方式如下：
@Override
 public void createNewSavepoint(ExecutionEnvironment env, String savepointPath, StateBackend stateBackend,
 ParameterTool config) {
    // 加载未修改max parallelism的savepoint
     String savepointOutputPath = config.get(EapSavepointConstants.EAP_SAVEPOINT_OUTPUT_PATH);
   // 新的max parallelism
     int maxParallelism = config.getInt(EapSavepointConstants.EAP_SAVEPOINT_MAX_PARALLELISM);
   
     try {
         ExistingSavepoint existingSavepoint = Savepoint.load(env, savepointPath, stateBackend);
         
        // 读取kafka相关state
         DataSet<Tuple2<KafkaTopicPartition, Long>> kafkaListState = existingSavepoint.readUnionState(
         OperatorUidAndNameConstants.KAFKA_SOURCE_UID, StateNameConstants.KAFKA_OFFSET_STATE_NAME,
         KafkaStateUtils.createTypeInformation(),
         KafkaStateUtils.createStateDescriptorSerializer(env.getConfig()));
         logger.info("Print kafka offset");
         kafkaListState.print();
        BootstrapTransformation<Tuple2<KafkaTopicPartition, Long>> kafkaTransformation = OperatorTransformation.bootstrapWith(kafkaListState).transform(new KafkaSourceBootstrapFunction());

         Savepoint.create(stateBackend, maxParallelism)
             .withOperator(OperatorUidAndNameConstants.KAFKA_SOURCE_UID, kafkaTransformation)
             .write(savepointOutputPath);
     } catch (IOException e) {
          logger.error("Savepoint load: " + e.getMessage());
          e.printStackTrace();
    } catch (Exception e) {
         logger.error("print state: " + e.getMessage());
         e.printStackTrace();
    }
 }

 

// KafkaStateUtils.java

public class KafkaStateUtils {
    /**
     * Creates state serializer for kafka topic partition to offset tuple.
     * Using of the explicit state serializer with KryoSerializer is needed because otherwise
     * users cannot use 'disableGenericTypes' properties with KafkaConsumer.
     * @param executionConfig
     * @return
     */
     public static TupleSerializer<Tuple2<KafkaTopicPartition, Long>> createStateDescriptorSerializer(
               ExecutionConfig executionConfig) {
         // explicit serializer will keep the compatibility with GenericTypeInformation
         // and allow to disableGenericTypes for users
         TypeSerializer<?>[] fieldSerializers = new TypeSerializer<?>[]{
                 new KryoSerializer<>(KafkaTopicPartition.class, executionConfig),
                 LongSerializer.INSTANCE
         };
        @SuppressWarnings("unchecked")
        Class<Tuple2<KafkaTopicPartition, Long>> tupleClass = (Class<Tuple2<KafkaTopicPartition, Long>>) (Class<?>) Tuple2.class;
        return new TupleSerializer<>(tupleClass, fieldSerializers);
     }


     public static TypeInformation<Tuple2<KafkaTopicPartition, Long>> createTypeInformation() {
         return TypeInformation.of(new TypeHint<Tuple2<KafkaTopicPartition, Long>>() {});
     }
}


// KafkaSourceBootstrapFunction.java
public class KafkaSourceBootstrapFunction extends StateBootstrapFunction<Tuple2<KafkaTopicPartition, Long>> {
           private ListState<Tuple2<KafkaTopicPartition, Long>> unionOffsetStates;
          @Override
          public void processElement(Tuple2<KafkaTopicPartition, Long> value, Context context) throws Exception {
              unionOffsetStates.add(value);
          }


         @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception { }
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            unionOffsetStates = context.getOperatorStateStore().getUnionListState(
                   new ListStateDescriptor<Tuple2<KafkaTopicPartition, Long>>(StateNameConstants.KAFKA_OFFSET_STATE_NAME,
                              KafkaStateUtils.createStateDescriptorSerializer(getRuntimeContext().getExecutionConfig())));
        }
}


以上是我的使用方式，请问是否可以有人给出一点使用建议？远程debug后发现只有`org.apache.flink.state.api.output.TaggedOperatorSubtaskState`没有从新的savepoint中反序列化出来。


Best regard!
clare