### Question:
OperatorChain implements StreamStatusMaintainer { public void prepareSnapshotPreBarrier(long checkpointId) throws Exception { final StreamOperator<?>[] operators = this.allOperators; for (int i = operators.length - 1; i >= 0; --i) { final StreamOperator<?> op = operators[i]; if (op != null) { op.prepareSnapshotPreBarrier(checkpointId); } } } OperatorChain(
			StreamTask<OUT, OP> containingTask,
			List<RecordWriter<SerializationDelegate<StreamRecord<OUT>>>> recordWriters); @VisibleForTesting  OperatorChain(
			StreamOperator<?>[] allOperators,
			RecordWriterOutput<?>[] streamOutputs,
			WatermarkGaugeExposingOutput<StreamRecord<OUT>> chainEntryPoint,
			OP headOperator); @Override StreamStatus getStreamStatus(); @Override void toggleStreamStatus(StreamStatus status); void broadcastCheckpointBarrier(long id, long timestamp, CheckpointOptions checkpointOptions); void broadcastCheckpointCancelMarker(long id); void prepareSnapshotPreBarrier(long checkpointId); void endInput(int inputId); RecordWriterOutput<?>[] getStreamOutputs(); StreamOperator<?>[] getAllOperators(); WatermarkGaugeExposingOutput<StreamRecord<OUT>> getChainEntryPoint(); void flushOutputs(); void releaseOutputs(); OP getHeadOperator(); int getChainLength();  }### Answer:
@Test public void testPrepareCheckpointPreBarrier() throws Exception { final AtomicInteger intRef = new AtomicInteger(); final OneInputStreamOperator<String, String> one = new ValidatingOperator(intRef, 0); final OneInputStreamOperator<String, String> two = new ValidatingOperator(intRef, 1); final OneInputStreamOperator<String, String> three = new ValidatingOperator(intRef, 2); final OperatorChain<?, ?> chain = setupOperatorChain(one, two, three); chain.prepareSnapshotPreBarrier(ValidatingOperator.CHECKPOINT_ID); assertEquals(3, intRef.get()); }