public LongRunningProcessStatus getDrainStatus(String workerId) {
        long startTime = System.nanoTime();
        LongRunningProcessStatus status = Optional.ofNullable(workerId).map(id ->
                Optional.ofNullable(drainOpStatusMap.get(id)).map(opStatus ->
                        switch (opStatus) {
                            case DrainCompleted ->
                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.SUCCESS);
                            case DrainInProgress ->
                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.RUNNING);
                            case DrainNotInProgress ->
                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.NOT_RUN);
                        }).orElse(
                        LongRunningProcessStatus.forError("Worker " + id + " not found in drain records")
                )
        ).orElse(
                new LongRunningProcessStatus()
        );
        log.info("Get drain status for worker {} - execution time: {} sec; returning status={}, error={}",
                workerId, NANOSECONDS.toSeconds (System.nanoTime() - startTime),
                status.status, status.lastError);
        return status;
    }
@Test
    public void testGetDrainStatus() throws Exception {
        // Clear the drain status map in the SchedulerManager; all other parameters are don't care for a clear.
        callGetDrainStatus(null, DrainOps.ClearDrainMap,
                SchedulerManager.DrainOpStatus.DrainCompleted);

        // Set up drain status for some fake workers.
        callGetDrainStatus("worker-1", DrainOps.SetDrainStatus,
                SchedulerManager.DrainOpStatus.DrainCompleted);
        callGetDrainStatus("worker-2", DrainOps.SetDrainStatus,
                SchedulerManager.DrainOpStatus.DrainInProgress);
        callGetDrainStatus("worker-3", DrainOps.SetDrainStatus,
                SchedulerManager.DrainOpStatus.DrainNotInProgress);

        // Get status; the status passed in a a don't care.
        LongRunningProcessStatus dStatus;
        dStatus = callGetDrainStatus("worker-0", DrainOps.GetDrainStatus,
                SchedulerManager.DrainOpStatus.DrainNotInProgress);
        Assert.assertTrue(dStatus.status == LongRunningProcessStatus.Status.ERROR);
        Assert.assertTrue(dStatus.lastError.matches("(.)+(not found)(.)+"));

        dStatus = callGetDrainStatus("worker-1", DrainOps.GetDrainStatus,
                SchedulerManager.DrainOpStatus.DrainNotInProgress);
        Assert.assertTrue(dStatus.status == LongRunningProcessStatus.Status.SUCCESS);
        Assert.assertTrue(dStatus.lastError.isEmpty());

        dStatus = callGetDrainStatus("worker-2", DrainOps.GetDrainStatus,
                SchedulerManager.DrainOpStatus.DrainNotInProgress);
        Assert.assertTrue(dStatus.status == LongRunningProcessStatus.Status.RUNNING);
        Assert.assertTrue(dStatus.lastError.isEmpty());

        dStatus = callGetDrainStatus("worker-3", DrainOps.GetDrainStatus,
                SchedulerManager.DrainOpStatus.DrainNotInProgress);
        Assert.assertTrue(dStatus.status == LongRunningProcessStatus.Status.NOT_RUN);
        Assert.assertTrue(dStatus.lastError.isEmpty());
    }