@Override
    public ApplicationLock lockApplication(OrchestratorContext context, ApplicationInstanceReference reference)
            throws UncheckedTimeoutException {

        Runnable onRegistryClose;

        // A multi-application operation, aka batch suspension, will first issue a probe
        // then a non-probe. With "large locks", the lock is not released in between -
        // no lock is taken on the non-probe. Instead, the release is done on the multi-application
        // context close.
        if (context.hasLock(reference)) {
            onRegistryClose = () -> {};
        } else {
            Runnable unlock = acquireLock(context, reference);
            if (context.registerLockAcquisition(reference, unlock)) {
                onRegistryClose = () -> {};
            } else {
                onRegistryClose = unlock;
            }
        }

        try {
            return new ZkApplicationLock(
                    this,
                    curator,
                    onRegistryClose,
                    reference,
                    context.isProbe(),
                    hostInfosCache);
        } catch (Throwable t) {
            // In case the constructor throws an exception.
            onRegistryClose.run();
            throw t;
        }
    }
@Test
    public void failing_to_get_lock_closes_SessionFailRetryLoop() {
        when(context.getTimeLeft()).thenReturn(Duration.ofMillis(100));
        ZkStatusService zkStatusService2 =
                new ZkStatusService(curator, mock(Metric.class), new TestTimer(), antiServiceMonitor);

        try (ApplicationLock lock = statusService
                .lockApplication(context, TestIds.APPLICATION_INSTANCE_REFERENCE)) {

            // must run in separate thread, since having 2 locks in the same thread fails
            CompletableFuture<Void> resultOfZkOperationAfterLockFailure = CompletableFuture.runAsync(() -> {
                try {
                    zkStatusService2.lockApplication(context, TestIds.APPLICATION_INSTANCE_REFERENCE);
                    fail("Both zookeeper host status services locked simultaneously for the same application instance");
                }
                catch (RuntimeException expected) { }
                finally {
                    killSession(curator.framework(), testingServer);
                }

                // Throws SessionFailedException if the SessionFailRetryLoop has not been closed.
                lock.getHostInfos().getOrNoRemarks(TestIds.HOST_NAME1);
            });

            assertThat(resultOfZkOperationAfterLockFailure, notHoldsException());
        }
    }