class OperatorEventValve {

	public void shutValve(long checkpointId) {
		synchronized (lock) {
			if (checkpointId == currentCheckpointId) {
				shut = true;
			} else {
				throw new IllegalStateException(String.format(
					"Cannot shut valve for non-prepared checkpoint. " +
					"Prepared checkpoint = %s, attempting-to-close checkpoint = %d",
					(currentCheckpointId == NO_CHECKPOINT ? "(none)" : String.valueOf(currentCheckpointId)),
					checkpointId));
			}
		}
	}

	public  OperatorEventValve(BiFunction<SerializedValue<OperatorEvent>, Integer, CompletableFuture<Acknowledge>> eventSender);

}