@Test public void testProcessTaskCheckResultPromoteUnsuccessfulParse() throws Exception { String snapshot = "snapshot1"; initSnapshotMetadata(snapshot, ProcessingStatus.UNINITIALIZED); SnapshotId snapshotId = _idManager.getSnapshotId(snapshot, _networkId).get(); NodeRolesId snapshotNodeRolesId = _idManager.getSnapshotNodeRolesId(_networkId, snapshotId); Main.getWorkMgr() .getStorage() .storeNodeRoles(NodeRolesData.builder().build(), snapshotNodeRolesId); assertFalse(_idManager.hasNetworkNodeRolesId(_networkId)); QueuedWork work1 = new QueuedWork( new WorkItem(NETWORK, snapshot), WorkDetails.builder() .setWorkType(WorkType.PARSING) .setNetworkId(_networkId) .setSnapshotId(_idManager.getSnapshotId(snapshot, _networkId).get()) .build()); doAction(new Action(ActionType.QUEUE, work1)); _workQueueMgr.processTaskCheckResult(work1, new Task(TaskStatus.TerminatedAbnormally, "Fake")); assertFalse(_idManager.hasNetworkNodeRolesId(_networkId)); }
public synchronized void processTaskCheckResult(QueuedWork work, Task task) throws Exception { switch (task.getStatus()) { case Unscheduled: case InProgress: work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); break; case TerminatedAbnormally: case TerminatedByUser: case TerminatedNormally: case RequeueFailure: { _queueIncompleteWork.delete(work); _queueCompletedWork.enque(work); work.setStatus(WorkStatusCode.fromTerminatedTaskStatus(task.getStatus())); work.recordTaskCheckResult(task); WorkItem wItem = work.getWorkItem(); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING) { ProcessingStatus status; if (task.getStatus() == TaskStatus.TerminatedNormally) { status = ProcessingStatus.PARSED; Main.getWorkMgr() .tryPromoteSnapshotNodeRoles(wDetails.getNetworkId(), wDetails.getSnapshotId()); } else { status = ProcessingStatus.PARSING_FAIL; } _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), status, task.getErrMessage()); } else if (wDetails.getWorkType() == WorkType.DATAPLANING) { if (task.getStatus() == TaskStatus.TerminatedAbnormally || task.getStatus() == TaskStatus.TerminatedByUser) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANING_FAIL, task.getErrMessage()); } else if (task.getStatus() == TaskStatus.TerminatedNormally) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANED, null); } } if (_blockingWork.contains(wItem.getId())) { _blockingWork.remove(wItem.getId()); List<QueuedWork> requeueWorks = new LinkedList<>(); for (QueuedWork incompleteWork : _queueIncompleteWork) { if (incompleteWork.getStatus() == WorkStatusCode.BLOCKED && wDetails.isOverlappingInput(incompleteWork.getDetails())) { requeueWorks.add(incompleteWork); } } for (QueuedWork requeueWork : requeueWorks) { _queueIncompleteWork.delete(requeueWork); requeueWork.setStatus(WorkStatusCode.UNASSIGNED); } for (QueuedWork requeueWork : requeueWorks) { try { boolean queued = queueUnassignedWork(requeueWork); if (!queued) { throw new BatfishException( "Failed to requeue previously blocked work " + requeueWork.getId()); } } catch (Exception e) { String stackTrace = Throwables.getStackTraceAsString(e); _logger.errorf("exception: %s\n", stackTrace); _queueIncompleteWork.enque(requeueWork); Task fakeTask = new Task( TaskStatus.RequeueFailure, String.format("Couldn't requeue after unblocking.\n%s", e.getMessage())); processTaskCheckResult(requeueWork, fakeTask); } } } } break; case Unknown: work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); break; case UnreachableOrBadResponse: { if (work.getLastTaskCheckResult().getStatus() == TaskStatus.UnreachableOrBadResponse) { work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); work.recordTaskCheckResult(task); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING || wDetails.getWorkType() == WorkType.DATAPLANING) { InitializationMetadata metadata = _snapshotMetadataManager.getInitializationMetadata( wDetails.getNetworkId(), wDetails.getSnapshotId()); if (wDetails.getWorkType() == WorkType.PARSING) { if (metadata.getProcessingStatus() != ProcessingStatus.PARSING) { _logger.errorf( "Unexpected status %s when parsing failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.UNINITIALIZED, task.getErrMessage()); } } else { if (metadata.getProcessingStatus() != ProcessingStatus.DATAPLANING) { _logger.errorf( "Unexpected status %s when dataplaning failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.PARSED, task.getErrMessage()); } } } } else { work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); } } break; default: throw new BatfishException( "Unhandled " + TaskStatus.class.getCanonicalName() + ": " + task.getStatus()); } }
WorkQueueMgr { public synchronized void processTaskCheckResult(QueuedWork work, Task task) throws Exception { switch (task.getStatus()) { case Unscheduled: case InProgress: work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); break; case TerminatedAbnormally: case TerminatedByUser: case TerminatedNormally: case RequeueFailure: { _queueIncompleteWork.delete(work); _queueCompletedWork.enque(work); work.setStatus(WorkStatusCode.fromTerminatedTaskStatus(task.getStatus())); work.recordTaskCheckResult(task); WorkItem wItem = work.getWorkItem(); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING) { ProcessingStatus status; if (task.getStatus() == TaskStatus.TerminatedNormally) { status = ProcessingStatus.PARSED; Main.getWorkMgr() .tryPromoteSnapshotNodeRoles(wDetails.getNetworkId(), wDetails.getSnapshotId()); } else { status = ProcessingStatus.PARSING_FAIL; } _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), status, task.getErrMessage()); } else if (wDetails.getWorkType() == WorkType.DATAPLANING) { if (task.getStatus() == TaskStatus.TerminatedAbnormally || task.getStatus() == TaskStatus.TerminatedByUser) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANING_FAIL, task.getErrMessage()); } else if (task.getStatus() == TaskStatus.TerminatedNormally) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANED, null); } } if (_blockingWork.contains(wItem.getId())) { _blockingWork.remove(wItem.getId()); List<QueuedWork> requeueWorks = new LinkedList<>(); for (QueuedWork incompleteWork : _queueIncompleteWork) { if (incompleteWork.getStatus() == WorkStatusCode.BLOCKED && wDetails.isOverlappingInput(incompleteWork.getDetails())) { requeueWorks.add(incompleteWork); } } for (QueuedWork requeueWork : requeueWorks) { _queueIncompleteWork.delete(requeueWork); requeueWork.setStatus(WorkStatusCode.UNASSIGNED); } for (QueuedWork requeueWork : requeueWorks) { try { boolean queued = queueUnassignedWork(requeueWork); if (!queued) { throw new BatfishException( "Failed to requeue previously blocked work " + requeueWork.getId()); } } catch (Exception e) { String stackTrace = Throwables.getStackTraceAsString(e); _logger.errorf("exception: %s\n", stackTrace); _queueIncompleteWork.enque(requeueWork); Task fakeTask = new Task( TaskStatus.RequeueFailure, String.format("Couldn't requeue after unblocking.\n%s", e.getMessage())); processTaskCheckResult(requeueWork, fakeTask); } } } } break; case Unknown: work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); break; case UnreachableOrBadResponse: { if (work.getLastTaskCheckResult().getStatus() == TaskStatus.UnreachableOrBadResponse) { work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); work.recordTaskCheckResult(task); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING || wDetails.getWorkType() == WorkType.DATAPLANING) { InitializationMetadata metadata = _snapshotMetadataManager.getInitializationMetadata( wDetails.getNetworkId(), wDetails.getSnapshotId()); if (wDetails.getWorkType() == WorkType.PARSING) { if (metadata.getProcessingStatus() != ProcessingStatus.PARSING) { _logger.errorf( "Unexpected status %s when parsing failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.UNINITIALIZED, task.getErrMessage()); } } else { if (metadata.getProcessingStatus() != ProcessingStatus.DATAPLANING) { _logger.errorf( "Unexpected status %s when dataplaning failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.PARSED, task.getErrMessage()); } } } } else { work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); } } break; default: throw new BatfishException( "Unhandled " + TaskStatus.class.getCanonicalName() + ": " + task.getStatus()); } } }
WorkQueueMgr { public synchronized void processTaskCheckResult(QueuedWork work, Task task) throws Exception { switch (task.getStatus()) { case Unscheduled: case InProgress: work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); break; case TerminatedAbnormally: case TerminatedByUser: case TerminatedNormally: case RequeueFailure: { _queueIncompleteWork.delete(work); _queueCompletedWork.enque(work); work.setStatus(WorkStatusCode.fromTerminatedTaskStatus(task.getStatus())); work.recordTaskCheckResult(task); WorkItem wItem = work.getWorkItem(); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING) { ProcessingStatus status; if (task.getStatus() == TaskStatus.TerminatedNormally) { status = ProcessingStatus.PARSED; Main.getWorkMgr() .tryPromoteSnapshotNodeRoles(wDetails.getNetworkId(), wDetails.getSnapshotId()); } else { status = ProcessingStatus.PARSING_FAIL; } _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), status, task.getErrMessage()); } else if (wDetails.getWorkType() == WorkType.DATAPLANING) { if (task.getStatus() == TaskStatus.TerminatedAbnormally || task.getStatus() == TaskStatus.TerminatedByUser) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANING_FAIL, task.getErrMessage()); } else if (task.getStatus() == TaskStatus.TerminatedNormally) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANED, null); } } if (_blockingWork.contains(wItem.getId())) { _blockingWork.remove(wItem.getId()); List<QueuedWork> requeueWorks = new LinkedList<>(); for (QueuedWork incompleteWork : _queueIncompleteWork) { if (incompleteWork.getStatus() == WorkStatusCode.BLOCKED && wDetails.isOverlappingInput(incompleteWork.getDetails())) { requeueWorks.add(incompleteWork); } } for (QueuedWork requeueWork : requeueWorks) { _queueIncompleteWork.delete(requeueWork); requeueWork.setStatus(WorkStatusCode.UNASSIGNED); } for (QueuedWork requeueWork : requeueWorks) { try { boolean queued = queueUnassignedWork(requeueWork); if (!queued) { throw new BatfishException( "Failed to requeue previously blocked work " + requeueWork.getId()); } } catch (Exception e) { String stackTrace = Throwables.getStackTraceAsString(e); _logger.errorf("exception: %s\n", stackTrace); _queueIncompleteWork.enque(requeueWork); Task fakeTask = new Task( TaskStatus.RequeueFailure, String.format("Couldn't requeue after unblocking.\n%s", e.getMessage())); processTaskCheckResult(requeueWork, fakeTask); } } } } break; case Unknown: work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); break; case UnreachableOrBadResponse: { if (work.getLastTaskCheckResult().getStatus() == TaskStatus.UnreachableOrBadResponse) { work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); work.recordTaskCheckResult(task); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING || wDetails.getWorkType() == WorkType.DATAPLANING) { InitializationMetadata metadata = _snapshotMetadataManager.getInitializationMetadata( wDetails.getNetworkId(), wDetails.getSnapshotId()); if (wDetails.getWorkType() == WorkType.PARSING) { if (metadata.getProcessingStatus() != ProcessingStatus.PARSING) { _logger.errorf( "Unexpected status %s when parsing failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.UNINITIALIZED, task.getErrMessage()); } } else { if (metadata.getProcessingStatus() != ProcessingStatus.DATAPLANING) { _logger.errorf( "Unexpected status %s when dataplaning failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.PARSED, task.getErrMessage()); } } } } else { work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); } } break; default: throw new BatfishException( "Unhandled " + TaskStatus.class.getCanonicalName() + ": " + task.getStatus()); } } WorkQueueMgr(BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager);   WorkQueueMgr(Type wqType, BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager); }
WorkQueueMgr { public synchronized void processTaskCheckResult(QueuedWork work, Task task) throws Exception { switch (task.getStatus()) { case Unscheduled: case InProgress: work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); break; case TerminatedAbnormally: case TerminatedByUser: case TerminatedNormally: case RequeueFailure: { _queueIncompleteWork.delete(work); _queueCompletedWork.enque(work); work.setStatus(WorkStatusCode.fromTerminatedTaskStatus(task.getStatus())); work.recordTaskCheckResult(task); WorkItem wItem = work.getWorkItem(); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING) { ProcessingStatus status; if (task.getStatus() == TaskStatus.TerminatedNormally) { status = ProcessingStatus.PARSED; Main.getWorkMgr() .tryPromoteSnapshotNodeRoles(wDetails.getNetworkId(), wDetails.getSnapshotId()); } else { status = ProcessingStatus.PARSING_FAIL; } _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), status, task.getErrMessage()); } else if (wDetails.getWorkType() == WorkType.DATAPLANING) { if (task.getStatus() == TaskStatus.TerminatedAbnormally || task.getStatus() == TaskStatus.TerminatedByUser) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANING_FAIL, task.getErrMessage()); } else if (task.getStatus() == TaskStatus.TerminatedNormally) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANED, null); } } if (_blockingWork.contains(wItem.getId())) { _blockingWork.remove(wItem.getId()); List<QueuedWork> requeueWorks = new LinkedList<>(); for (QueuedWork incompleteWork : _queueIncompleteWork) { if (incompleteWork.getStatus() == WorkStatusCode.BLOCKED && wDetails.isOverlappingInput(incompleteWork.getDetails())) { requeueWorks.add(incompleteWork); } } for (QueuedWork requeueWork : requeueWorks) { _queueIncompleteWork.delete(requeueWork); requeueWork.setStatus(WorkStatusCode.UNASSIGNED); } for (QueuedWork requeueWork : requeueWorks) { try { boolean queued = queueUnassignedWork(requeueWork); if (!queued) { throw new BatfishException( "Failed to requeue previously blocked work " + requeueWork.getId()); } } catch (Exception e) { String stackTrace = Throwables.getStackTraceAsString(e); _logger.errorf("exception: %s\n", stackTrace); _queueIncompleteWork.enque(requeueWork); Task fakeTask = new Task( TaskStatus.RequeueFailure, String.format("Couldn't requeue after unblocking.\n%s", e.getMessage())); processTaskCheckResult(requeueWork, fakeTask); } } } } break; case Unknown: work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); break; case UnreachableOrBadResponse: { if (work.getLastTaskCheckResult().getStatus() == TaskStatus.UnreachableOrBadResponse) { work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); work.recordTaskCheckResult(task); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING || wDetails.getWorkType() == WorkType.DATAPLANING) { InitializationMetadata metadata = _snapshotMetadataManager.getInitializationMetadata( wDetails.getNetworkId(), wDetails.getSnapshotId()); if (wDetails.getWorkType() == WorkType.PARSING) { if (metadata.getProcessingStatus() != ProcessingStatus.PARSING) { _logger.errorf( "Unexpected status %s when parsing failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.UNINITIALIZED, task.getErrMessage()); } } else { if (metadata.getProcessingStatus() != ProcessingStatus.DATAPLANING) { _logger.errorf( "Unexpected status %s when dataplaning failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.PARSED, task.getErrMessage()); } } } } else { work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); } } break; default: throw new BatfishException( "Unhandled " + TaskStatus.class.getCanonicalName() + ": " + task.getStatus()); } } WorkQueueMgr(BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager);   WorkQueueMgr(Type wqType, BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager); synchronized List<QueuedWork> getCompletedWork(
      NetworkId networkId, SnapshotId snapshotId); synchronized long getLength(QueueType qType); synchronized JSONObject getStatusJson(); synchronized QueuedWork getMatchingWork(WorkItem workItem, QueueType qType); synchronized QueuedWork getWork(UUID workId); @Nullable synchronized QueuedWork getWorkForAssignment(); @Nonnull synchronized List<QueuedWork> getWorkForChecking(); synchronized List<QueuedWork> listIncompleteWork(
      NetworkId networkId, @Nullable SnapshotId snapshotId, @Nullable WorkType workType); synchronized void makeWorkUnassigned(QueuedWork work); synchronized void markAssignmentError(QueuedWork work); synchronized void markAssignmentFailure(QueuedWork work); synchronized void markAssignmentSuccess(QueuedWork work, String assignedWorker); synchronized void processTaskCheckResult(QueuedWork work, Task task); synchronized boolean queueUnassignedWork(QueuedWork work); }
WorkQueueMgr { public synchronized void processTaskCheckResult(QueuedWork work, Task task) throws Exception { switch (task.getStatus()) { case Unscheduled: case InProgress: work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); break; case TerminatedAbnormally: case TerminatedByUser: case TerminatedNormally: case RequeueFailure: { _queueIncompleteWork.delete(work); _queueCompletedWork.enque(work); work.setStatus(WorkStatusCode.fromTerminatedTaskStatus(task.getStatus())); work.recordTaskCheckResult(task); WorkItem wItem = work.getWorkItem(); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING) { ProcessingStatus status; if (task.getStatus() == TaskStatus.TerminatedNormally) { status = ProcessingStatus.PARSED; Main.getWorkMgr() .tryPromoteSnapshotNodeRoles(wDetails.getNetworkId(), wDetails.getSnapshotId()); } else { status = ProcessingStatus.PARSING_FAIL; } _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), status, task.getErrMessage()); } else if (wDetails.getWorkType() == WorkType.DATAPLANING) { if (task.getStatus() == TaskStatus.TerminatedAbnormally || task.getStatus() == TaskStatus.TerminatedByUser) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANING_FAIL, task.getErrMessage()); } else if (task.getStatus() == TaskStatus.TerminatedNormally) { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.DATAPLANED, null); } } if (_blockingWork.contains(wItem.getId())) { _blockingWork.remove(wItem.getId()); List<QueuedWork> requeueWorks = new LinkedList<>(); for (QueuedWork incompleteWork : _queueIncompleteWork) { if (incompleteWork.getStatus() == WorkStatusCode.BLOCKED && wDetails.isOverlappingInput(incompleteWork.getDetails())) { requeueWorks.add(incompleteWork); } } for (QueuedWork requeueWork : requeueWorks) { _queueIncompleteWork.delete(requeueWork); requeueWork.setStatus(WorkStatusCode.UNASSIGNED); } for (QueuedWork requeueWork : requeueWorks) { try { boolean queued = queueUnassignedWork(requeueWork); if (!queued) { throw new BatfishException( "Failed to requeue previously blocked work " + requeueWork.getId()); } } catch (Exception e) { String stackTrace = Throwables.getStackTraceAsString(e); _logger.errorf("exception: %s\n", stackTrace); _queueIncompleteWork.enque(requeueWork); Task fakeTask = new Task( TaskStatus.RequeueFailure, String.format("Couldn't requeue after unblocking.\n%s", e.getMessage())); processTaskCheckResult(requeueWork, fakeTask); } } } } break; case Unknown: work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); break; case UnreachableOrBadResponse: { if (work.getLastTaskCheckResult().getStatus() == TaskStatus.UnreachableOrBadResponse) { work.setStatus(WorkStatusCode.UNASSIGNED); work.clearAssignment(); work.recordTaskCheckResult(task); WorkDetails wDetails = work.getDetails(); if (wDetails.getWorkType() == WorkType.PARSING || wDetails.getWorkType() == WorkType.DATAPLANING) { InitializationMetadata metadata = _snapshotMetadataManager.getInitializationMetadata( wDetails.getNetworkId(), wDetails.getSnapshotId()); if (wDetails.getWorkType() == WorkType.PARSING) { if (metadata.getProcessingStatus() != ProcessingStatus.PARSING) { _logger.errorf( "Unexpected status %s when parsing failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.UNINITIALIZED, task.getErrMessage()); } } else { if (metadata.getProcessingStatus() != ProcessingStatus.DATAPLANING) { _logger.errorf( "Unexpected status %s when dataplaning failed for %s", metadata.getProcessingStatus(), wDetails.getSnapshotId()); } else { _snapshotMetadataManager.updateInitializationStatus( wDetails.getNetworkId(), wDetails.getSnapshotId(), ProcessingStatus.PARSED, task.getErrMessage()); } } } } else { work.setStatus(WorkStatusCode.ASSIGNED); work.recordTaskCheckResult(task); } } break; default: throw new BatfishException( "Unhandled " + TaskStatus.class.getCanonicalName() + ": " + task.getStatus()); } } WorkQueueMgr(BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager);   WorkQueueMgr(Type wqType, BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager); synchronized List<QueuedWork> getCompletedWork(
      NetworkId networkId, SnapshotId snapshotId); synchronized long getLength(QueueType qType); synchronized JSONObject getStatusJson(); synchronized QueuedWork getMatchingWork(WorkItem workItem, QueueType qType); synchronized QueuedWork getWork(UUID workId); @Nullable synchronized QueuedWork getWorkForAssignment(); @Nonnull synchronized List<QueuedWork> getWorkForChecking(); synchronized List<QueuedWork> listIncompleteWork(
      NetworkId networkId, @Nullable SnapshotId snapshotId, @Nullable WorkType workType); synchronized void makeWorkUnassigned(QueuedWork work); synchronized void markAssignmentError(QueuedWork work); synchronized void markAssignmentFailure(QueuedWork work); synchronized void markAssignmentSuccess(QueuedWork work, String assignedWorker); synchronized void processTaskCheckResult(QueuedWork work, Task task); synchronized boolean queueUnassignedWork(QueuedWork work);  }