void pollAndExecute(Worker worker) {
        Boolean discoveryOverride =
                Optional.ofNullable(
                                PropertyFactory.getBoolean(
                                        worker.getTaskDefName(), OVERRIDE_DISCOVERY, null))
                        .orElseGet(
                                () ->
                                        PropertyFactory.getBoolean(
                                                ALL_WORKERS, OVERRIDE_DISCOVERY, false));

        if (eurekaClient != null
                && !eurekaClient.getInstanceRemoteStatus().equals(InstanceStatus.UP)
                && !discoveryOverride) {
            LOGGER.debug("Instance is NOT UP in discovery - will not poll");
            return;
        }

        if (worker.paused()) {
            MetricsContainer.incrementTaskPausedCount(worker.getTaskDefName());
            LOGGER.debug("Worker {} has been paused. Not polling anymore!", worker.getClass());
            return;
        }

        String taskType = worker.getTaskDefName();
        PollingSemaphore pollingSemaphore = getPollingSemaphore(taskType);

        int slotsToAcquire = pollingSemaphore.availableSlots();
        if (slotsToAcquire <= 0 || !pollingSemaphore.acquireSlots(slotsToAcquire)) {
            return;
        }
        int acquiredTasks = 0;
        try {
            String domain =
                    Optional.ofNullable(PropertyFactory.getString(taskType, DOMAIN, null))
                            .orElseGet(
                                    () ->
                                            Optional.ofNullable(
                                                            PropertyFactory.getString(
                                                                    ALL_WORKERS, DOMAIN, null))
                                                    .orElse(taskToDomain.get(taskType)));

            LOGGER.debug("Polling task of type: {} in domain: '{}'", taskType, domain);

            List<Task> tasks =
                    MetricsContainer.getPollTimer(taskType)
                            .record(
                                    () ->
                                            taskClient.batchPollTasksInDomain(
                                                    taskType,
                                                    domain,
                                                    worker.getIdentity(),
                                                    slotsToAcquire,
                                                    worker.getBatchPollTimeoutInMS()));
            acquiredTasks = tasks.size();
            for (Task task : tasks) {
                if (Objects.nonNull(task) && StringUtils.isNotBlank(task.getTaskId())) {
                    MetricsContainer.incrementTaskPollCount(taskType, 1);
                    LOGGER.debug(
                            "Polled task: {} of type: {} in domain: '{}', from worker: {}",
                            task.getTaskId(),
                            taskType,
                            domain,
                            worker.getIdentity());

                    CompletableFuture<Task> taskCompletableFuture =
                            CompletableFuture.supplyAsync(
                                    () -> processTask(task, worker, pollingSemaphore),
                                    executorService);

                    if (task.getResponseTimeoutSeconds() > 0 && worker.leaseExtendEnabled()) {
                        ScheduledFuture<?> leaseExtendFuture =
                                leaseExtendExecutorService.scheduleWithFixedDelay(
                                        extendLease(task, taskCompletableFuture),
                                        Math.round(
                                                task.getResponseTimeoutSeconds()
                                                        * LEASE_EXTEND_DURATION_FACTOR),
                                        Math.round(
                                                task.getResponseTimeoutSeconds()
                                                        * LEASE_EXTEND_DURATION_FACTOR),
                                        TimeUnit.SECONDS);
                        leaseExtendMap.put(task.getTaskId(), leaseExtendFuture);
                    }

                    taskCompletableFuture.whenComplete(this::finalizeTask);
                } else {
                    // no task was returned in the poll, release the permit
                    pollingSemaphore.complete(1);
                }
            }
        } catch (Exception e) {
            MetricsContainer.incrementTaskPollErrorCount(worker.getTaskDefName(), e);
            LOGGER.error("Error when polling for tasks", e);
        }

        // immediately release unused permits
        pollingSemaphore.complete(slotsToAcquire - acquiredTasks);
    }
Manages the threadpool used by the workers for execution and server communication (polling and
task update).
client/src/main/java/com/netflix/conductor/client/automator/TaskPollExecutor.java