public function runTransaction(callable $callable, array $options = [])
    {
        $options += [
            'maxRetries' => self::MAX_RETRIES,
            'begin' => [],
            'commit' => [],
            'rollback' => []
        ];

        $retryableErrors = [
            AbortedException::class
        ];

        $delayFn = function () {
            return [
                'seconds' => 0,
                'nanos' => 0
            ];
        };

        $retryFn = function (\Exception $e) use ($retryableErrors) {
            return in_array(get_class($e), $retryableErrors);
        };

        // Track the Transaction ID outside the retry function.
        // If the transaction is retried after an abort, the previous transaction
        // must be provided to the subsequent `beginTransaction` rpc.
        // It also provides a convenient indication to the user whether the
        // transaction is retried or not.
        $transactionId = null;

        $retry = new Retry($options['maxRetries'], $delayFn, $retryFn);

        return $retry->execute(function (
            callable $callable,
            array $options
        ) use (&$transactionId) {
            $database = $this->databaseName($this->projectId, $this->database);

            $beginTransaction = $this->connection->beginTransaction(array_filter([
                'database' => $database,
                'retryTransaction' => $transactionId
            ]) + $options['begin']);

            $transactionId = $beginTransaction['transaction'];

            $transaction = new Transaction(
                $this->connection,
                $this->valueMapper,
                $database,
                $transactionId
            );

            try {
                $res = $callable($transaction);

                if (!$transaction->writer()->isEmpty()) {
                    $transaction->writer()->commit([
                        'transaction' => $transactionId
                    ] + $options['commit']);
                } else {
                    // trigger rollback if no writes exist.
                    $transaction->writer()->rollback($options['rollback']);
                }

                return $res;
            } catch (\Exception $e) {
                $transaction->writer()->rollback($options['rollback']);

                throw $e;
            }
        }, [
            $callable,
            $options
        ]);
    }