import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
  if (checkpointCoder == null) {
    // no checkpoint coder available in this source
    return;
  }

  OperatorStateStore stateStore = context.getOperatorStateStore();
  CoderTypeInformation<
      KV<? extends UnboundedSource<OutputT, CheckpointMarkT>, CheckpointMarkT>>
      typeInformation = (CoderTypeInformation) new CoderTypeInformation<>(checkpointCoder);
  stateForCheckpoint = stateStore.getOperatorState(
      new ListStateDescriptor<>(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME,
          typeInformation.createSerializer(new ExecutionConfig())));

  if (context.isRestored()) {
    isRestored = true;
    LOG.info("Having restore state in the UnbounedSourceWrapper.");
  } else {
    LOG.info("No restore state for UnbounedSourceWrapper.");
  }
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@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(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.");
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Before
public void beforeTest() throws Exception {

	OperatorStateStore mockStore = Mockito.mock(OperatorStateStore.class);
	FunctionInitializationContext mockContext = Mockito.mock(FunctionInitializationContext.class);
	Mockito.when(mockContext.getOperatorStateStore()).thenReturn(mockStore);
	Mockito.when(mockStore.getSerializableListState(any(String.class))).thenReturn(null);

	source = new RMQTestSource();
	source.initializeState(mockContext);
	source.open(config);

	messageId = 0;
	generateCorrelationIds = true;

	sourceThread = new Thread(new Runnable() {
		@Override
		public void run() {
			try {
				source.run(new DummySourceContext());
			} catch (Exception e) {
				exception = e;
			}
		}
	});
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {

	if (broadcast) {
		this.counterPartitions = context
				.getOperatorStateStore()
				.getUnionListState(new ListStateDescriptor<>("counter_partitions", IntSerializer.INSTANCE));
	} else {
		this.counterPartitions = context
				.getOperatorStateStore()
				.getListState(new ListStateDescriptor<>("counter_partitions", IntSerializer.INSTANCE));
	}

	if (context.isRestored()) {
		for (int v : counterPartitions.get()) {
			counter += v;
		}
		checkCorrectRestore[getRuntimeContext().getIndexOfThisSubtask()] = counter;
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	Preconditions.checkState(this.checkpointedState == null,
		"The " + getClass().getSimpleName() + " has already been initialized.");

	this.checkpointedState = context.getOperatorStateStore().getListState(
		new ListStateDescriptor<>(
			"from-elements-state",
			IntSerializer.INSTANCE
		)
	);

	if (context.isRestored()) {
		List<Integer> retrievedStates = new ArrayList<>();
		for (Integer entry : this.checkpointedState.get()) {
			retrievedStates.add(entry);
		}

		// given that the parallelism of the function is 1, we can only have 1 state
		Preconditions.checkArgument(retrievedStates.size() == 1,
			getClass().getSimpleName() + " retrieved invalid state.");

		this.numElementsToSkip = retrievedStates.get(0);
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {

	OperatorStateStore stateStore = context.getOperatorStateStore();
	offsetsStateForCheckpoint = stateStore.getSerializableListState(DefaultOperatorStateBackend.DEFAULT_OPERATOR_STATE_NAME);

	if (context.isRestored()) {
		restoreToOffset = new HashMap<>();
		for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : offsetsStateForCheckpoint.get()) {
			restoreToOffset.put(kafkaOffset.f0, kafkaOffset.f1);
		}

		LOG.info("Setting restore state in the FlinkKafkaConsumer.");
		if (LOG.isDebugEnabled()) {
			LOG.debug("Using the following offsets: {}", restoreToOffset);
		}
	} else {
		LOG.info("No restore state for FlinkKafkaConsumer.");
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
public void initializeState(FunctionInitializationContext context) throws Exception {
    LOG.info("initializeState...");
    TypeInformation<Tuple2<LogstoreShardMeta, String>> shardsStateTypeInfo = new TupleTypeInfo<Tuple2<LogstoreShardMeta, String>>(
            TypeInformation.of(LogstoreShardMeta.class),
            TypeInformation.of(String.class));

    cursorStateForCheckpoint = context.getOperatorStateStore().getUnionListState(
            new ListStateDescriptor(curcorStateStoreName, shardsStateTypeInfo));

    if (context.isRestored()) {
        if (cursorsToRestore == null) {
            cursorsToRestore = new HashMap<LogstoreShardMeta, String>();
            for (Tuple2<LogstoreShardMeta, String> cursor : cursorStateForCheckpoint.get()) {
                LOG.info("initializeState, project: {}, logstore: {}, shard: {}, checkpoint: {}", logProject, logStore, cursor.f0.toString(), cursor.f1);
                cursorsToRestore.put(cursor.f0, cursor.f1);
                if (consumerGroupName != null && logClient != null) {
                    logClient.updateCheckpoint(logProject, logStore, consumerGroupName, "flinkTask-" + getRuntimeContext().getIndexOfThisSubtask() + "Of" + getRuntimeContext().getNumberOfParallelSubtasks(), cursor.f0.getShardId(), cursor.f1);
                }
            }

            LOG.info("Setting restore state in the FlinkLogConsumer. Using the following offsets: {}",
                    cursorsToRestore);
        }
    }
    else {
        LOG.info("No restore state for FlinkLogConsumer.");
    }
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
    Preconditions.checkArgument(this.restoredBucketStates == null, "The operator has already been initialized.");

    try {
        initFileSystem();
    } catch (IOException e) {
        LOG.error("Error while creating FileSystem when initializing the state of the TODBucketingSink.", e);
        throw new RuntimeException("Error while creating FileSystem when initializing the state of the TODBucketingSink.", e);
    }

    if (this.refTruncate == null) {
        this.refTruncate = reflectTruncate(fs);
    }

    OperatorStateStore stateStore = context.getOperatorStateStore();
    restoredBucketStates = stateStore.getSerializableListState("bucket-states");

    int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
    if (context.isRestored()) {
        LOG.info("Restoring state for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);

        for (State<T> recoveredState : restoredBucketStates.get()) {
            handleRestoredBucketState(recoveredState);
            if (LOG.isDebugEnabled()) {
                LOG.debug("{} idx {} restored {}", getClass().getSimpleName(), subtaskIndex, recoveredState);
            }
        }
    } else {
        LOG.info("No state to restore for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);
    }
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	if (semantic != Semantic.NONE && !((StreamingRuntimeContext) this.getRuntimeContext()).isCheckpointingEnabled()) {
		LOG.warn("Using {} semantic, but checkpointing is not enabled. Switching to {} semantic.", semantic, Semantic.NONE);
		semantic = Semantic.NONE;
	}

	nextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(
		NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
	transactionalIdsGenerator = new TransactionalIdsGenerator(
		getRuntimeContext().getTaskName(),
		getRuntimeContext().getIndexOfThisSubtask(),
		getRuntimeContext().getNumberOfParallelSubtasks(),
		kafkaProducersPoolSize,
		SAFE_SCALE_DOWN_FACTOR);

	if (semantic != Semantic.EXACTLY_ONCE) {
		nextTransactionalIdHint = null;
	} else {
		ArrayList<NextTransactionalIdHint> transactionalIdHints = Lists.newArrayList(nextTransactionalIdHintState.get());
		if (transactionalIdHints.size() > 1) {
			throw new IllegalStateException(
				"There should be at most one next transactional id hint written by the first subtask");
		} else if (transactionalIdHints.size() == 0) {
			nextTransactionalIdHint = new NextTransactionalIdHint(0, 0);

			// this means that this is either:
			// (1) the first execution of this application
			// (2) previous execution has failed before first checkpoint completed
			//
			// in case of (2) we have to abort all previous transactions
			abortTransactions(transactionalIdsGenerator.generateIdsToAbort());
		} else {
			nextTransactionalIdHint = transactionalIdHints.get(0);
		}
	}

	super.initializeState(context);
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	Preconditions.checkArgument(this.restoredBucketStates == null,
		"The " + getClass().getSimpleName() + " has already been initialized.");

	try {
		initFileSystem();
	} catch (IOException e) {
		LOG.error("Error while creating FileSystem when initializing the state of the RollingSink.", e);
		throw new RuntimeException("Error while creating FileSystem when initializing the state of the RollingSink.", e);
	}

	if (this.refTruncate == null) {
		this.refTruncate = reflectTruncate(fs);
	}

	OperatorStateStore stateStore = context.getOperatorStateStore();
	restoredBucketStates = stateStore.getSerializableListState("rolling-states");

	int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
	if (context.isRestored()) {
		LOG.info("Restoring state for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);

		for (BucketState bucketState : restoredBucketStates.get()) {
			handleRestoredBucketState(bucketState);
		}

		if (LOG.isDebugEnabled()) {
			LOG.debug("{} (taskIdx= {}) restored {}", getClass().getSimpleName(), subtaskIndex, bucketState);
		}
	} else {
		LOG.info("No state to restore for the {} (taskIdx= {}).", getClass().getSimpleName(), subtaskIndex);
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	Preconditions.checkArgument(this.restoredBucketStates == null, "The operator has already been initialized.");

	try {
		initFileSystem();
	} catch (IOException e) {
		LOG.error("Error while creating FileSystem when initializing the state of the BucketingSink.", e);
		throw new RuntimeException("Error while creating FileSystem when initializing the state of the BucketingSink.", e);
	}

	if (this.refTruncate == null) {
		this.refTruncate = reflectTruncate(fs);
	}

	OperatorStateStore stateStore = context.getOperatorStateStore();
	restoredBucketStates = stateStore.getSerializableListState("bucket-states");

	int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
	if (context.isRestored()) {
		LOG.info("Restoring state for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);

		for (State<T> recoveredState : restoredBucketStates.get()) {
			handleRestoredBucketState(recoveredState);
			if (LOG.isDebugEnabled()) {
				LOG.debug("{} idx {} restored {}", getClass().getSimpleName(), subtaskIndex, recoveredState);
			}
		}
	} else {
		LOG.info("No state to restore for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	pojoClass = getRuntimeContext().getUserCodeClassLoader().loadClass(POJO_NAME);

	fieldA = pojoClass.getDeclaredField("a");
	fieldA.setAccessible(true);

	if (hasBField) {
		fieldB = pojoClass.getDeclaredField("b");
		fieldB.setAccessible(true);
	}

	if (keyed) {
		keyedValueState = context.getKeyedStateStore().getState(
			new ValueStateDescriptor<>("keyedValueState", (Class<Object>) pojoClass));
		keyedMapState = context.getKeyedStateStore().getMapState(
			new MapStateDescriptor<>("keyedMapState", (Class<Object>) pojoClass, (Class<Object>) pojoClass));
		keyedListState = context.getKeyedStateStore().getListState(
			new ListStateDescriptor<>("keyedListState", (Class<Object>) pojoClass));

		ReduceFunction<Object> reduceFunction = new FirstValueReducer<>();
		keyedReducingState = context.getKeyedStateStore().getReducingState(
			new ReducingStateDescriptor<>("keyedReducingState", reduceFunction, (Class<Object>) pojoClass));
	} else {
		partitionableListState = context.getOperatorStateStore().getListState(
			new ListStateDescriptor<>("partitionableListState", (Class<Object>) pojoClass));
		unionListState = context.getOperatorStateStore().getUnionListState(
			new ListStateDescriptor<>("unionListState", (Class<Object>) pojoClass));
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	ValueStateDescriptor<Integer> descriptor =
		new ValueStateDescriptor<>(
			"seenCountState",
			TypeInformation.of(new TypeHint<Integer>() {}),
			0);
	alreadySeen = context.getKeyedStateStore().getState(descriptor);
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {

	Preconditions.checkState(this.checkpointedState == null,
		"The " + getClass().getSimpleName() + " has already been initialized.");

	this.checkpointedState = context.getOperatorStateStore().getListState(
		new ListStateDescriptor<>(
			"stateful-sequence-source-state",
			LongSerializer.INSTANCE
		)
	);

	this.valuesToEmit = new ArrayDeque<>();
	if (context.isRestored()) {
		// upon restoring

		for (Long v : this.checkpointedState.get()) {
			this.valuesToEmit.add(v);
		}
	} else {
		// the first time the job is executed

		final int stepSize = getRuntimeContext().getNumberOfParallelSubtasks();
		final int taskIdx = getRuntimeContext().getIndexOfThisSubtask();
		final long congruence = start + taskIdx;

		long totalNoOfElements = Math.abs(end - start + 1);
		final int baseSize = safeDivide(totalNoOfElements, stepSize);
		final int toCollect = (totalNoOfElements % stepSize > taskIdx) ? baseSize + 1 : baseSize;

		for (long collected = 0; collected < toCollect; collected++) {
			this.valuesToEmit.add(collected * stepSize + congruence);
		}
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	Preconditions.checkState(this.checkpointedState == null,
		"The " + getClass().getSimpleName() + " has already been initialized.");

	this.checkpointedState = context
		.getOperatorStateStore()
		.getSerializableListState("message-acknowledging-source-state");

	this.idsForCurrentCheckpoint = new HashSet<>(64);
	this.pendingCheckpoints = new ArrayDeque<>();
	this.idsProcessedButNotAcknowledged = new HashSet<>();

	if (context.isRestored()) {
		LOG.info("Restoring state for the {}.", getClass().getSimpleName());

		List<SerializedCheckpointData[]> retrievedStates = new ArrayList<>();
		for (SerializedCheckpointData[] entry : this.checkpointedState.get()) {
			retrievedStates.add(entry);
		}

		// given that the parallelism of the function is 1, we can only have at most 1 state
		Preconditions.checkArgument(retrievedStates.size() == 1,
			getClass().getSimpleName() + " retrieved invalid state.");

		pendingCheckpoints = SerializedCheckpointData.toDeque(retrievedStates.get(0), idSerializer);
		// build a set which contains all processed ids. It may be used to check if we have
		// already processed an incoming message.
		for (Tuple2<Long, Set<UId>> checkpoint : pendingCheckpoints) {
			idsProcessedButNotAcknowledged.addAll(checkpoint.f1);
		}
	} else {
		LOG.info("No state to restore for the {}.", getClass().getSimpleName());
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	serializableListState = context
			.getOperatorStateStore()
			.getListState(new ListStateDescriptor<>("test-state", IntSerializer.INSTANCE));
	if (context.isRestored()) {
		Iterator<Integer> integers = serializableListState.get().iterator();
		int act = integers.next();
		Assert.assertEquals(42, act);
		Assert.assertFalse(integers.hasNext());
		wasRestored = true;
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	Preconditions.checkArgument(this.restoredBucketStates == null,
		"The " + getClass().getSimpleName() + " has already been initialized.");

	initFileSystem();

	if (this.refTruncate == null) {
		this.refTruncate = reflectTruncate(fs);
	}

	OperatorStateStore stateStore = context.getOperatorStateStore();
	restoredBucketStates = stateStore.getSerializableListState("rolling-states");

	int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
	if (context.isRestored()) {
		LOG.info("Restoring state for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);

		for (BucketState bucketState : restoredBucketStates.get()) {
			handleRestoredBucketState(bucketState);
		}

		if (LOG.isDebugEnabled()) {
			LOG.debug("{} (taskIdx= {}) restored {}", getClass().getSimpleName(), subtaskIndex, bucketState);
		}
	} else {
		LOG.info("No state to restore for the {} (taskIdx= {}).", getClass().getSimpleName(), subtaskIndex);
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
	Preconditions.checkArgument(this.restoredBucketStates == null, "The operator has already been initialized.");

	initFileSystem();

	if (this.refTruncate == null) {
		this.refTruncate = reflectTruncate(fs);
	}

	OperatorStateStore stateStore = context.getOperatorStateStore();
	restoredBucketStates = stateStore.getSerializableListState("bucket-states");

	int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
	if (context.isRestored()) {
		LOG.info("Restoring state for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);

		for (State<T> recoveredState : restoredBucketStates.get()) {
			handleRestoredBucketState(recoveredState);
			if (LOG.isDebugEnabled()) {
				LOG.debug("{} idx {} restored {}", getClass().getSimpleName(), subtaskIndex, recoveredState);
			}
		}
	} else {
		LOG.info("No state to restore for the {} (taskIdx={}).", getClass().getSimpleName(), subtaskIndex);
	}
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
@Override
public void initializeState(FunctionInitializationContext functionInitializationContext) throws Exception {
  //nothing to initialize, as in flight documents are completely flushed during checkpointing
}
 

import org.apache.flink.runtime.state.FunctionInitializationContext; //导入依赖的package包/类
public void initializeState(FunctionInitializationContext context) throws Exception {
    // do nothing
}