code: @Override
  	public void initializeState(FunctionInitializationContext context) throws Exception {
  		TypeInformation<Tuple2<StreamShardMetadata, SequenceNumber>> shardsStateTypeInfo = new TupleTypeInfo<>(
  			TypeInformation.of(StreamShardMetadata.class),
  			TypeInformation.of(SequenceNumber.class));
  
  		sequenceNumsStateForCheckpoint = context.getOperatorStateStore().getUnionListState(
  			new ListStateDescriptor<>(sequenceNumsStateStoreName, shardsStateTypeInfo));
  
  		if (context.isRestored()) {
  			if (sequenceNumsToRestore == null) {
  				sequenceNumsToRestore = new HashMap<>();
  				for (Tuple2<StreamShardMetadata, SequenceNumber> kinesisSequenceNumber : sequenceNumsStateForCheckpoint.get()) {
  					sequenceNumsToRestore.put(
  						// we wrap the restored metadata inside an equivalence wrapper that checks only stream name and shard id,
  						// so that if a shard had been closed (due to a Kinesis reshard operation, for example) since
  						// the savepoint and has a different metadata than what we last stored,
  						// we will still be able to match it in sequenceNumsToRestore. Please see FLINK-8484 for details.
  						new StreamShardMetadata.EquivalenceWrapper(kinesisSequenceNumber.f0),
  						kinesisSequenceNumber.f1);
  				}
  
  				LOG.info("Setting restore state in the FlinkKinesisConsumer. Using the following offsets: {}",
  					sequenceNumsToRestore);
  			}
  		} else {
  			LOG.info("No restore state for FlinkKinesisConsumer.");
  		}
  	}
docs: ------------------------------------------------------------------------