class OperatorSnapshotFutures {

	public void cancel() throws Exception {
		List<Tuple2<Future<? extends StateObject>, String>> pairs = new ArrayList<>();
		pairs.add(new Tuple2<>(getKeyedStateManagedFuture(), "managed keyed"));
		pairs.add(new Tuple2<>(getKeyedStateRawFuture(), "managed operator"));
		pairs.add(new Tuple2<>(getOperatorStateManagedFuture(), "raw keyed"));
		pairs.add(new Tuple2<>(getOperatorStateRawFuture(), "raw operator"));
		pairs.add(new Tuple2<>(getInputChannelStateFuture(), "input channel"));
		pairs.add(new Tuple2<>(getResultSubpartitionStateFuture(), "result subpartition"));
		try (Closer closer = Closer.create()) {
			for (Tuple2<Future<? extends StateObject>, String> pair : pairs) {
				closer.register(() -> {
					try {
						discardStateFuture(pair.f0);
					} catch (Exception e) {
						throw new RuntimeException(String.format("Could not properly cancel %s state future", pair.f1), e);
					}
				});
			}
		}
	}

	public  OperatorSnapshotFutures();
	public  OperatorSnapshotFutures(
			@Nonnull RunnableFuture<SnapshotResult<KeyedStateHandle>> keyedStateManagedFuture,
			@Nonnull RunnableFuture<SnapshotResult<KeyedStateHandle>> keyedStateRawFuture,
			@Nonnull RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateManagedFuture,
			@Nonnull RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateRawFuture,
			@Nonnull Future<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>> inputChannelStateFuture,
			@Nonnull Future<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>> resultSubpartitionStateFuture);

}