code: @Override
    public void resetOperation(Operation op)
    {
      LOGGER.debug("Reset Loader");
  
      if (op.equals(_op))
      {
        //no-op
        return;
      }
  
      LOGGER.debug("Operation is changing from {} to {}", _op, op);
      _op = op;
  
      if (_stage != null)
      {
        try
        {
          queuePut(_stage);
        }
        catch (InterruptedException ex)
        {
          LOGGER.error(_stage.getId(), ex);
        }
      }
  
      _stage = new BufferStage(this, _op, _csvFileBucketSize, _csvFileSize);
  
    }
docs: If operation changes, existing stage needs to be scheduled for processing.