if (taskIsRunningOrStaging(taskStatus)) {
<BUG>log.info("Adding task to restart list. Block: " + getName() + " Status: " + taskStatus);
</BUG>
KafkaScheduler.restartTasks(fetchTaskInfo());
<BUG>return null;
}</BUG>
try {
OfferRequirement offerReq = getOfferRequirement(fetchTaskInfo());
List<TaskID> newPendingTasks = new ArrayList<TaskID>();
for (TaskRequirement taskRequirement : offerReq.getTaskRequirements()) {

if (taskIsRunningOrStaging(taskStatus)) {
log.info("Adding task to restart list. Block: " + getName() + " Status: " + taskStatus.get());
KafkaScheduler.restartTasks(fetchTaskInfo());
return Optional.empty();
}
try {
OfferRequirement offerReq = getOfferRequirement(fetchTaskInfo());
List<TaskID> newPendingTasks = new ArrayList<TaskID>();
for (TaskRequirement taskRequirement : offerReq.getTaskRequirements()) {
