docs: @Override
  	public StreamOperatorStateContext streamOperatorStateContext(
  		@Nonnull OperatorID operatorID,
  		@Nonnull String operatorClassName,
  		@Nonnull KeyContext keyContext,
  		@Nullable TypeSerializer<?> keySerializer,
  		@Nonnull CloseableRegistry streamTaskCloseableRegistry,
  		@Nonnull MetricGroup metricGroup) throws Exception {
  
  		TaskInfo taskInfo = environment.getTaskInfo();
  		OperatorSubtaskDescriptionText operatorSubtaskDescription =
  			new OperatorSubtaskDescriptionText(
  				operatorID,
  				operatorClassName,
  				taskInfo.getIndexOfThisSubtask(),
  				taskInfo.getNumberOfParallelSubtasks());
  
  		final String operatorIdentifierText = operatorSubtaskDescription.toString();
  
  		final PrioritizedOperatorSubtaskState prioritizedOperatorSubtaskStates =
  			taskStateManager.prioritizedOperatorState(operatorID);
  
  		AbstractKeyedStateBackend<?> keyedStatedBackend = null;
  		OperatorStateBackend operatorStateBackend = null;
  		CloseableIterable<KeyGroupStatePartitionStreamProvider> rawKeyedStateInputs = null;
  		CloseableIterable<StatePartitionStreamProvider> rawOperatorStateInputs = null;
  		InternalTimeServiceManager<?> timeServiceManager;
  
  		try {
  
  			// -------------- Keyed State Backend --------------
  			keyedStatedBackend = keyedStatedBackend(
  				keySerializer,
  				operatorIdentifierText,
  				prioritizedOperatorSubtaskStates,
  				streamTaskCloseableRegistry,
  				metricGroup);
  
  			// -------------- Operator State Backend --------------
  			operatorStateBackend = operatorStateBackend(
  				operatorIdentifierText,
  				prioritizedOperatorSubtaskStates,
  				streamTaskCloseableRegistry);
  
  			// -------------- Raw State Streams --------------
  			rawKeyedStateInputs = rawKeyedStateInputs(
  				prioritizedOperatorSubtaskStates.getPrioritizedRawKeyedState().iterator());
  			streamTaskCloseableRegistry.registerCloseable(rawKeyedStateInputs);
  
  			rawOperatorStateInputs = rawOperatorStateInputs(
  				prioritizedOperatorSubtaskStates.getPrioritizedRawOperatorState().iterator());
  			streamTaskCloseableRegistry.registerCloseable(rawOperatorStateInputs);
  
  			// -------------- Internal Timer Service Manager --------------
  			timeServiceManager = internalTimeServiceManager(keyedStatedBackend, keyContext, rawKeyedStateInputs);
  
  			// -------------- Preparing return value --------------
  
  			return new StreamOperatorStateContextImpl(
  				prioritizedOperatorSubtaskStates.isRestored(),
  				operatorStateBackend,
  				keyedStatedBackend,
  				timeServiceManager,
  				rawOperatorStateInputs,
  				rawKeyedStateInputs);
  		} catch (Exception ex) {
  
  			// cleanup if something went wrong before results got published.
  			if (keyedStatedBackend != null) {
  				if (streamTaskCloseableRegistry.unregisterCloseable(keyedStatedBackend)) {
  					IOUtils.closeQuietly(keyedStatedBackend);
  				}
  				// release resource (e.g native resource)
  				keyedStatedBackend.dispose();
  			}
  
  			if (operatorStateBackend != null) {
  				if (streamTaskCloseableRegistry.unregisterCloseable(operatorStateBackend)) {
  					IOUtils.closeQuietly(operatorStateBackend);
  				}
  				operatorStateBackend.dispose();
  			}
  
  			if (streamTaskCloseableRegistry.unregisterCloseable(rawKeyedStateInputs)) {
  				IOUtils.closeQuietly(rawKeyedStateInputs);
  			}
  
  			if (streamTaskCloseableRegistry.unregisterCloseable(rawOperatorStateInputs)) {
  				IOUtils.closeQuietly(rawOperatorStateInputs);
  			}
  
  			throw new Exception("Exception while creating StreamOperatorStateContext.", ex);
  		}
  	}