def acquire(self, key: str, blocking: bool=True, timeout: float=None, metadata: Any=None,
                on_before_lock: LockEventListener=lambda key: None,
                on_lock_already_locked: LockEventListener=lambda key: None,
                lock_poll_interval_generator: Callable[[int], float]=DEFAULT_LOCK_POLL_INTERVAL_GENERATOR) \
            -> Optional[ConnectedConsulLockInformation]:
        """
        Acquires a Consul lock.
        :param key: the lock key
        :param blocking: whether to block and wait for the lock
        :param timeout: timeout in seconds
        :param metadata: metadata to add to the lock information. Must be parsable by default JSON encode/decoder
        :param on_before_lock: event listener to be called before attempt to acquire lock
        :param on_lock_already_locked: event listener to be called when an attempt to acquire a lock has failed as the
        lock is already locked
        :param lock_poll_interval_generator: generator of the interval between Consul lock polls where the first
        argument is the attempt number (starting at 1)
        :return: information about the lock if acquired, else `None` if not acquired and not blocking
        :raises InvalidKeyError: raised if the given key is not valid
        :raises LockAcquireTimeoutError: raised if times out waiting for the lock
        """
        ConsulLockManager.validate_key(key)

        logger.debug("Creating Consul session...")
        session_id = self.consul_client.session.create(
            lock_delay=0, ttl=self.session_ttl_in_seconds, behavior="delete")
        self._acquiring_session_ids.add(session_id)
        logger.info(f"Created session with ID: {session_id}")
        start_time = monotonic()

        @timeout_decorator.timeout(timeout, timeout_exception=LockAcquireTimeoutError)
        def _acquire() -> Optional[ConsulLockInformation]:
            i = 1
            while True:
                logger.debug("Going to acquire lock")
                seconds_to_lock = monotonic() - start_time
                on_before_lock(key)
                lock_information = self._acquire_lock(key, session_id, seconds_to_lock, metadata)
                if lock_information is not None:
                    logger.debug("Acquired lock!")
                    return lock_information
                else:
                    on_lock_already_locked(key)
                    if not blocking:
                        logger.debug("Could not acquire lock (already locked) and not blocking")
                        return None
                    else:
                        logger.debug("Could not acquire lock (already locked)")
                interval = lock_poll_interval_generator(i)
                logger.debug(f"Sleeping for {interval}s")
                sleep(interval)
                i += 1

        lock_information = _acquire()
        self._acquiring_session_ids.remove(session_id)

        if lock_information is None:
            self.consul_client.session.destroy(session_id=session_id)
            logger.info(f"Destroyed session (did not acquire the lock)")

        return lock_information