code: @Override
  	public void dispose() throws Exception {
  
  		Exception exception = null;
  
  		StreamTask<?, ?> containingTask = getContainingTask();
  		CloseableRegistry taskCloseableRegistry = containingTask != null ?
  			containingTask.getCancelables() :
  			null;
  
  		try {
  			if (taskCloseableRegistry == null ||
  				taskCloseableRegistry.unregisterCloseable(operatorStateBackend)) {
  				operatorStateBackend.close();
  			}
  		} catch (Exception e) {
  			exception = e;
  		}
  
  		try {
  			if (taskCloseableRegistry == null ||
  				taskCloseableRegistry.unregisterCloseable(keyedStateBackend)) {
  				keyedStateBackend.close();
  			}
  		} catch (Exception e) {
  			exception = ExceptionUtils.firstOrSuppressed(e, exception);
  		}
  
  		try {
  			if (operatorStateBackend != null) {
  				operatorStateBackend.dispose();
  			}
  		} catch (Exception e) {
  			exception = ExceptionUtils.firstOrSuppressed(e, exception);
  		}
  
  		try {
  			if (keyedStateBackend != null) {
  				keyedStateBackend.dispose();
  			}
  		} catch (Exception e) {
  			exception = ExceptionUtils.firstOrSuppressed(e, exception);
  		}
  
  		if (exception != null) {
  			throw exception;
  		}
  	}
docs: This method is called at the very end of the operator's life, both in the case of a successful
  completion of the operation, and in the case of a failure and canceling.
  
  <p>This method is expected to make a thorough effort to release all resources
  that the operator has acquired.