void pollAndExecute(WorkflowSystemTask systemTask, String queueName) {
        if (!isRunning()) {
            LOGGER.debug(
                    "{} stopped. Not polling for task: {}", getClass().getSimpleName(), systemTask);
            return;
        }

        ExecutionConfig executionConfig = getExecutionConfig(queueName);
        SemaphoreUtil semaphoreUtil = executionConfig.getSemaphoreUtil();
        ExecutorService executorService = executionConfig.getExecutorService();
        String taskName = QueueUtils.getTaskType(queueName);

        int messagesToAcquire = semaphoreUtil.availableSlots();

        try {
            if (messagesToAcquire <= 0 || !semaphoreUtil.acquireSlots(messagesToAcquire)) {
                // no available slots, do not poll
                Monitors.recordSystemTaskWorkerPollingLimited(queueName);
                return;
            }

            LOGGER.debug("Polling queue: {} with {} slots acquired", queueName, messagesToAcquire);

            List<String> polledTaskIds = queueDAO.pop(queueName, messagesToAcquire, 200);

            Monitors.recordTaskPoll(queueName);
            LOGGER.debug("Polling queue:{}, got {} tasks", queueName, polledTaskIds.size());

            if (polledTaskIds.size() > 0) {
                // Immediately release unused slots when number of messages acquired is less than
                // acquired slots
                if (polledTaskIds.size() < messagesToAcquire) {
                    semaphoreUtil.completeProcessing(messagesToAcquire - polledTaskIds.size());
                }

                for (String taskId : polledTaskIds) {
                    if (StringUtils.isNotBlank(taskId)) {
                        LOGGER.debug(
                                "Task: {} from queue: {} being sent to the workflow executor",
                                taskId,
                                queueName);
                        Monitors.recordTaskPollCount(queueName, 1);

                        executionService.ackTaskReceived(taskId);

                        CompletableFuture<Void> taskCompletableFuture =
                                CompletableFuture.runAsync(
                                        () -> asyncSystemTaskExecutor.execute(systemTask, taskId),
                                        executorService);

                        // release permit after processing is complete
                        taskCompletableFuture.whenComplete(
                                (r, e) -> semaphoreUtil.completeProcessing(1));
                    } else {
                        semaphoreUtil.completeProcessing(1);
                    }
                }
            } else {
                // no task polled, release permit
                semaphoreUtil.completeProcessing(messagesToAcquire);
            }
        } catch (Exception e) {
            // release the permit if exception is thrown during polling, because the thread would
            // not be busy
            semaphoreUtil.completeProcessing(messagesToAcquire);
            Monitors.recordTaskPollError(taskName, e.getClass().getSimpleName());
            LOGGER.error("Error polling system task in queue:{}", queueName, e);
        }
    }
The worker that polls and executes an async system task.
core/src/main/java/com/netflix/conductor/core/execution/tasks/SystemTaskWorker.java