long processedInputPositions = 0;
long outputDataSize = 0;
long outputPositions = 0;
boolean fullyBlocked = true;
Set<BlockedReason> blockedReasons = new HashSet<>();
<BUG>Map<Integer, OperatorStats> operatorToStats = new HashMap<>();
</BUG>
for (TaskInfo taskInfo : taskInfos) {
TaskState taskState = taskInfo.getTaskStatus().getState();
if (taskState.isDone()) {

long processedInputPositions = 0;
long outputDataSize = 0;
long outputPositions = 0;
boolean fullyBlocked = true;
Set<BlockedReason> blockedReasons = new HashSet<>();
Map<String, OperatorStats> operatorToStats = new HashMap<>();
for (TaskInfo taskInfo : taskInfos) {
TaskState taskState = taskInfo.getTaskStatus().getState();
if (taskState.isDone()) {
