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

	nextTransactionalIdHintState.clear();
	// To avoid duplication only first subtask keeps track of next transactional id hint. Otherwise all of the
	// subtasks would write exactly same information.
	if (getRuntimeContext().getIndexOfThisSubtask() == 0 && semantic == Semantic.EXACTLY_ONCE) {
		checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be set for EXACTLY_ONCE");
		long nextFreeTransactionalId = nextTransactionalIdHint.nextFreeTransactionalId;

		// If we scaled up, some (unknown) subtask must have created new transactional ids from scratch. In that
		// case we adjust nextFreeTransactionalId by the range of transactionalIds that could be used for this
		// scaling up.
		if (getRuntimeContext().getNumberOfParallelSubtasks() > nextTransactionalIdHint.lastParallelism) {
			nextFreeTransactionalId += getRuntimeContext().getNumberOfParallelSubtasks() * kafkaProducersPoolSize;
		}

		nextTransactionalIdHintState.add(new NextTransactionalIdHint(
			getRuntimeContext().getNumberOfParallelSubtasks(),
			nextFreeTransactionalId));
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
	// check for asynchronous errors and fail the checkpoint if necessary
	checkErroneous();

	if (flushOnCheckpoint) {
		// flushing is activated: We need to wait until pendingRecords is 0
		flush();
		synchronized (pendingRecordsLock) {
			if (pendingRecords != 0) {
				throw new IllegalStateException("Pending record count must be zero at this point: " + pendingRecords);
			}

			// if the flushed requests has errors, we should propagate it also and fail the checkpoint
			checkErroneous();
		}
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
	Preconditions.checkNotNull(restoredBucketStates,
		"The " + getClass().getSimpleName() + " has not been properly initialized.");

	int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();

	if (isWriterOpen) {
		bucketState.currentFile = currentPartPath.toString();
		bucketState.currentFileValidLength = writer.flush();
	}

	synchronized (bucketState.pendingFilesPerCheckpoint) {
		bucketState.pendingFilesPerCheckpoint.put(context.getCheckpointId(), bucketState.pendingFiles);
	}
	bucketState.pendingFiles = new ArrayList<>();

	restoredBucketStates.clear();
	restoredBucketStates.add(bucketState);

	if (LOG.isDebugEnabled()) {
		LOG.debug("{} (taskIdx={}) checkpointed {}.", getClass().getSimpleName(), subtaskIdx, bucketState);
	}
}
 

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

	counterPartitions.clear();

	checkCorrectSnapshot[getRuntimeContext().getIndexOfThisSubtask()] = counter;

	int div = counter / NUM_PARTITIONS;
	int mod = counter % NUM_PARTITIONS;

	for (int i = 0; i < NUM_PARTITIONS; ++i) {
		int partitionValue = div;
		if (mod > 0) {
			--mod;
			++partitionValue;
		}
		counterPartitions.add(partitionValue);
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
	// this is like the pre-commit of a 2-phase-commit transaction
	// we are ready to commit and remember the transaction

	checkState(currentTransactionHolder != null, "bug: no transaction object when performing state snapshot");

	long checkpointId = context.getCheckpointId();
	LOG.debug("{} - checkpoint {} triggered, flushing transaction '{}'", name(), context.getCheckpointId(), currentTransactionHolder);

	preCommit(currentTransactionHolder.handle);
	pendingCommitTransactions.put(checkpointId, currentTransactionHolder);
	LOG.debug("{} - stored pending transactions {}", name(), pendingCommitTransactions);

	currentTransactionHolder = beginTransactionInternal();
	LOG.debug("{} - started new transaction '{}'", name(), currentTransactionHolder);

	state.clear();
	state.add(new State<>(
		this.currentTransactionHolder,
		new ArrayList<>(pendingCommitTransactions.values()),
		userContext));
}
 

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

	if (LOG.isDebugEnabled()) {
		LOG.debug("{} checkpointing: Messages: {}, checkpoint id: {}, timestamp: {}",
			idsForCurrentCheckpoint, context.getCheckpointId(), context.getCheckpointTimestamp());
	}

	pendingCheckpoints.addLast(new Tuple2<>(context.getCheckpointId(), idsForCurrentCheckpoint));
	idsForCurrentCheckpoint = new HashSet<>(64);

	this.checkpointedState.clear();
	this.checkpointedState.add(SerializedCheckpointData.fromDeque(pendingCheckpoints, idSerializer));
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
	Preconditions.checkNotNull(restoredBucketStates,
		"The " + getClass().getSimpleName() + " has not been properly initialized.");

	int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
	
	if (isWriterOpen) {
		bucketState.currentFile = currentPartPath.toString();
		bucketState.currentFileValidLength = writer.flush();
	}

	synchronized (bucketState.pendingFilesPerCheckpoint) {
		bucketState.pendingFilesPerCheckpoint.put(context.getCheckpointId(), bucketState.pendingFiles);
	}
	bucketState.pendingFiles = new ArrayList<>();

	restoredBucketStates.clear();
	restoredBucketStates.add(bucketState);

	if (LOG.isDebugEnabled()) {
		LOG.debug("{} (taskIdx={}) checkpointed {}.", getClass().getSimpleName(), subtaskIdx, bucketState);
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
  do {
    try {
      flushDocumentBuffer();
    } catch (IOException e) {
      //if the request fails, that's fine, just retry on the next iteration
    }
  } while (! documentBuffer.isEmpty());
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    Preconditions.checkNotNull(restoredBucketStates, "The operator has not been properly initialized.");

    restoredBucketStates.clear();

    synchronized (state.bucketStates) {
        int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();

        for (Map.Entry<String, BucketState<T>> bucketStateEntry : state.bucketStates.entrySet()) {
            BucketState<T> bucketState = bucketStateEntry.getValue();

            if (bucketState.isWriterOpen) {
                bucketState.currentFileValidLength = bucketState.writer.flush();
            }

            synchronized (bucketState.pendingFilesPerCheckpoint) {
                bucketState.pendingFilesPerCheckpoint.put(context.getCheckpointId(), bucketState.pendingFiles);
            }
            bucketState.pendingFiles = new ArrayList<>();
        }
        restoredBucketStates.add(state);

        if (LOG.isDebugEnabled()) {
            LOG.debug("{} idx {} checkpointed {}.", getClass().getSimpleName(), subtaskIdx, state);
        }
    }
}
 

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

	if (flushOnCheckpoint) {
		do {
			bulkProcessor.flush();
			checkErrorAndRethrow();
		} while (numPendingRequests.get() != 0);
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
	isFlushed = false;

	super.snapshotState(ctx);

	// if the snapshot implementation doesn't wait until all pending records are flushed, we should fail the test
	if (flushOnCheckpoint && !isFlushed) {
		throw new RuntimeException("Flushing is enabled; snapshots should be blocked until all pending records are flushed");
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
	// check for asynchronous errors and fail the checkpoint if necessary
	checkAndPropagateAsyncError();

	flushSync();
	if (producer.getOutstandingRecordsCount() > 0) {
		throw new IllegalStateException(
			"Number of outstanding records must be zero at this point: " + producer.getOutstandingRecordsCount());
	}

	// if the flushed requests has errors, we should propagate it also and fail the checkpoint
	checkAndPropagateAsyncError();
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
	isFlushed = false;

	super.snapshotState(context);

	// if the snapshot implementation doesn't wait until all pending records are flushed, we should fail the test
	if (!isFlushed) {
		throw new RuntimeException("Flushing is enabled; snapshots should be blocked until all pending records are flushed");
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
	Preconditions.checkNotNull(restoredBucketStates, "The operator has not been properly initialized.");

	restoredBucketStates.clear();

	synchronized (state.bucketStates) {
		int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();

		for (Map.Entry<String, BucketState<T>> bucketStateEntry : state.bucketStates.entrySet()) {
			BucketState<T> bucketState = bucketStateEntry.getValue();

			if (bucketState.isWriterOpen) {
				bucketState.currentFileValidLength = bucketState.writer.flush();
			}

			synchronized (bucketState.pendingFilesPerCheckpoint) {
				bucketState.pendingFilesPerCheckpoint.put(context.getCheckpointId(), bucketState.pendingFiles);
			}
			bucketState.pendingFiles = new ArrayList<>();
		}
		restoredBucketStates.add(state);

		if (LOG.isDebugEnabled()) {
			LOG.debug("{} idx {} checkpointed {}.", getClass().getSimpleName(), subtaskIdx, state);
		}
	}
}
 

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

	this.checkpointedState.clear();
	for (Long v : this.valuesToEmit) {
		this.checkpointedState.add(v);
	}
}
 

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

	this.checkpointedState.clear();
	this.checkpointedState.add(this.numElementsEmitted);
}
 

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

	this.checkpointedState.clear();
	this.checkpointedState.add(this.globalModificationTime);

	if (LOG.isDebugEnabled()) {
		LOG.debug("{} checkpointed {}.", getClass().getSimpleName(), globalModificationTime);
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
	if (flushOnCheckpoint) {
		// flushing is activated: We need to wait until pendingRecords is 0
		flush();
		synchronized (pendingRecordsLock) {
			if (pendingRecords != 0) {
				throw new IllegalStateException("Pending record count must be zero at this point: " + pendingRecords);
			}
			// pending records count is 0. We can now confirm the checkpoint
		}
	}
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
@Override
public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
	// call the actual snapshot state
	super.snapshotState(ctx);
	// notify test that snapshotting has been done
	snapshottingFinished.set(true);
}
 

import org.apache.flink.runtime.state.FunctionSnapshotContext; //导入依赖的package包/类
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    if(logProducer != null) {
        logProducer.flush();
        Thread.sleep(logProducer.getProducerConfig().packageTimeoutInMS);
    }
}