private Runnable acquireLock(OrchestratorContext context, ApplicationInstanceReference reference)
            throws UncheckedTimeoutException {
        ApplicationId applicationId = OrchestratorUtil.toApplicationId(reference);
        String app = applicationId.application().value() + "." + applicationId.instance().value();
        Map<String, String> dimensions = Map.of(
                "tenantName", applicationId.tenant().value(),
                "applicationId", applicationId.toFullString(),
                "app", app);
        Metric.Context metricContext = cachedContexts.computeIfAbsent(dimensions, metric::createContext);

        Duration duration = context.getTimeLeft();
        String lockPath = applicationInstanceLock2Path(reference);
        Lock lock = new Lock(lockPath, curator);

        Instant startTime = timer.currentTime();
        Instant acquireEndTime;
        boolean lockAcquired = false;
        try {
            lock.acquire(duration);
            lockAcquired = true;
        } finally {
            acquireEndTime = timer.currentTime();
            double seconds = durationInSeconds(startTime, acquireEndTime);
            // TODO: These metrics are redundant with Lock's metrics
            metric.set(ConfigServerMetrics.ORCHESTRATOR_LOCK_ACQUIRE_LATENCY.baseName(), seconds, metricContext);
            metric.set(ConfigServerMetrics.ORCHESTRATOR_LOCK_ACQUIRED.baseName(), lockAcquired ? 1 : 0, metricContext);

            metric.add(ConfigServerMetrics.ORCHESTRATOR_LOCK_ACQUIRE.baseName(), 1, metricContext);
            String acquireResultMetricName = lockAcquired
                    ? ConfigServerMetrics.ORCHESTRATOR_LOCK_ACQUIRE_SUCCESS.baseName()
                    : ConfigServerMetrics.ORCHESTRATOR_LOCK_ACQUIRE_TIMEOUT.baseName();
            metric.add(acquireResultMetricName, 1, metricContext);
        }

        CriticalRegion inaccessibleDuperModelRegion = antiServiceMonitor
                .disallowDuperModelLockAcquisition(ZkStatusService.class.getSimpleName() + " application lock");

        return () -> {
            try {
                lock.close();
            } catch (RuntimeException e) {
                // We may want to avoid logging some exceptions that may be expected, like when session expires.
                log.log(Level.WARNING,
                        "Failed to close application lock for " +
                                ZkStatusService.class.getSimpleName() + ", will ignore and continue",
                        e);
            }

            inaccessibleDuperModelRegion.close();

            Instant lockReleasedTime = timer.currentTime();
            double seconds = durationInSeconds(acquireEndTime, lockReleasedTime);
            metric.set(ConfigServerMetrics.ORCHESTRATOR_LOCK_HOLD_LATENCY.baseName(), seconds, metricContext);
        };
    }
1) locks the status service for an application instance.
 2) fails all operations in this thread when the session is lost,
    since session loss might cause the lock to be lost.
    Since it only fails operations in this thread,
    all operations depending on a lock, including the locking itself, must be done in this thread.
    Note that since it is the thread that fails, all status operations in this thread will fail
    even if they're not supposed to be guarded by this lock
    (i.e. the request is for another applicationInstanceReference)
orchestrator/src/main/java/com/yahoo/vespa/orchestrator/status/ZkStatusService.java