code: private void registerCleanUpTimer(Context ctx, long rowTime, boolean leftRow) throws IOException {
  		if (leftRow) {
  			long cleanUpTime = rowTime + leftRelativeSize + minCleanUpInterval + allowedLateness + 1;
  			registerTimer(ctx, cleanUpTime);
  			rightTimerState.update(cleanUpTime);
  		} else {
  			long cleanUpTime = rowTime + rightRelativeSize + minCleanUpInterval + allowedLateness + 1;
  			registerTimer(ctx, cleanUpTime);
  			leftTimerState.update(cleanUpTime);
  		}
  	}
docs: Register a timer for cleaning up rows in a specified time.
  
  @param ctx the context to register timer
  @param rowTime time for the input row
  @param leftRow whether this row comes from the left stream