log.info("Starting block: " + getName() + " with status: " + Block.getStatus(this));
if (!isPending()) {
log.warn("Block is not pending.  start() should not be called.");
<BUG>return null;
}
TaskStatus taskStatus = fetchTaskStatus();
</BUG>
if (taskIsRunningOrStaging(taskStatus)) {
<BUG>log.info("Adding task to restart list. Block: " + getName() + " Status: " + taskStatus);
</BUG>
KafkaScheduler.restartTasks(fetchTaskInfo());

log.info("Starting block: " + getName() + " with status: " + Block.getStatus(this));
if (!isPending()) {
log.warn("Block is not pending.  start() should not be called.");
return Optional.empty();
}
Optional<TaskStatus> taskStatus = fetchTaskStatus();
if (taskIsRunningOrStaging(taskStatus)) {
log.info("Adding task to restart list. Block: " + getName() + " Status: " + taskStatus.get());
