code: @Override
    public void close() throws IOException {
      try {
        if (consumer != null && consumer instanceof Closeable) {
          // Call close in a new transaction.
          // TODO (terence): Actually need to coordinates with other flowlets to drain the queue.
          TransactionContext txContext = dataFabricFacade.createTransactionManager();
          txContext.start();
          try {
            ((Closeable) consumer).close();
            txContext.finish();
          } catch (TransactionFailureException e) {
            LOG.warn("Fail to commit transaction when closing consumer.");
            txContext.abort();
          }
        }
      } catch (Exception e) {
        LOG.warn("Fail to close queue consumer.", e);
      }
      consumer = null;
    }
docs: Close the current consumer if there is one.