code: @Override
  	public final void initializeState(FunctionInitializationContext context) throws Exception {
  
  		OperatorStateStore stateStore = context.getOperatorStateStore();
  
  		ListState<Tuple2<KafkaTopicPartition, Long>> oldRoundRobinListState =
  			stateStore.getSerializableListState(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME);
  
  		this.unionOffsetStates = stateStore.getUnionListState(new ListStateDescriptor<>(
  				OFFSETS_STATE_NAME,
  				TypeInformation.of(new TypeHint<Tuple2<KafkaTopicPartition, Long>>() {})));
  
  		if (context.isRestored() && !restoredFromOldState) {
  			restoredState = new TreeMap<>(new KafkaTopicPartition.Comparator());
  
  			// migrate from 1.2 state, if there is any
  			for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : oldRoundRobinListState.get()) {
  				restoredFromOldState = true;
  				unionOffsetStates.add(kafkaOffset);
  			}
  			oldRoundRobinListState.clear();
  
  			if (restoredFromOldState && discoveryIntervalMillis != PARTITION_DISCOVERY_DISABLED) {
  				throw new IllegalArgumentException(
  					"Topic / partition discovery cannot be enabled if the job is restored from a savepoint from Flink 1.2.x.");
  			}
  
  			// populate actual holder for restored state
  			for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : unionOffsetStates.get()) {
  				restoredState.put(kafkaOffset.f0, kafkaOffset.f1);
  			}
  
  			LOG.info("Setting restore state in the FlinkKafkaConsumer: {}", restoredState);
  		} else {
  			LOG.info("No restore state for FlinkKafkaConsumer.");
  		}
  	}
docs: ------------------------------------------------------------------------