public function consume($message, array $headers = [])
    {
        try {
            $this->transactionManager->beginTransaction();
        } catch (BeginException $e) {
            throw new ConsumerException($e->getMessage(), $e->getCode(), $e);
        }

        try {
            $this->consumer->consume($message, $headers);
            $this->transactionManager->commit();
        } catch (\Exception $e) {
            $this->transactionManager->rollback();
            throw $e;
        }
    }