@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 locks_are_exclusive() throws Exception {
        ZkStatusService zkStatusService2 =
                new ZkStatusService(curator, mock(Metric.class), new TestTimer(), antiServiceMonitor);

        final CompletableFuture<Void> lockedSuccessfullyFuture;
        try (ApplicationLock lock = statusService
                .lockApplication(context, TestIds.APPLICATION_INSTANCE_REFERENCE)) {

            lockedSuccessfullyFuture = CompletableFuture.runAsync(() -> {
                try (ApplicationLock lock2 = zkStatusService2
                        .lockApplication(context, TestIds.APPLICATION_INSTANCE_REFERENCE))
                {
                }
            });

            try {
                lockedSuccessfullyFuture.get(3, TimeUnit.SECONDS);
                fail("Both zookeeper host status services locked simultaneously for the same application instance");
            } catch (TimeoutException ignored) {
            }
        }

        lockedSuccessfullyFuture.get(1, TimeUnit.MINUTES);
    }