class KeyedStateInputFormat extends RichInputFormat<OUT, KeyGroupRangeInputSplit> implements KeyContext {

	@Override
	public KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits) throws IOException {
		final int maxParallelism = operatorState.getMaxParallelism();

		final List<KeyGroupRange> keyGroups = sortedKeyGroupRanges(minNumSplits, maxParallelism);

		return CollectionUtil.mapWithIndex(
			keyGroups,
			(keyGroupRange, index) -> createKeyGroupRangeInputSplit(
				operatorState,
				maxParallelism,
				keyGroupRange,
				index)
		).toArray(KeyGroupRangeInputSplit[]::new);
	}

	public  KeyedStateInputFormat(
		OperatorState operatorState,
		StateBackend stateBackend,
		TypeInformation<K> keyType,
		KeyedStateReaderFunction<K, OUT> userFunction);

}