@Test public void queueUnassignedWorkDuplicate() throws Exception { String snapshot = "snapshot1"; initSnapshotMetadata(snapshot, ProcessingStatus.UNINITIALIZED); QueuedWork work1 = new QueuedWork( new WorkItem(NETWORK, snapshot), WorkDetails.builder() .setWorkType(WorkType.UNKNOWN) .setNetworkId(_networkId) .setSnapshotId(_idManager.getSnapshotId(snapshot, _networkId).get()) .build()); _workQueueMgr.queueUnassignedWork(work1); _thrown.expect(BatfishException.class); _thrown.expectMessage("Duplicate work item"); _workQueueMgr.queueUnassignedWork(work1); }
public synchronized boolean queueUnassignedWork(QueuedWork work) throws Exception { QueuedWork previouslyQueuedWork = getWork(work.getId()); if (previouslyQueuedWork != null) { throw new BatfishException("Duplicate work item"); } WorkDetails wDetails = work.getDetails(); cleanUpInitMetaDataIfNeeded(work.getDetails().getNetworkId(), wDetails.getSnapshotId()); if (work.getDetails().isDifferential()) { cleanUpInitMetaDataIfNeeded( work.getDetails().getNetworkId(), wDetails.getReferenceSnapshotId()); } switch (work.getDetails().getWorkType()) { case PARSING: return queueParsingWork(work); case DATAPLANING: return queueDataplaningWork(work); case INDEPENDENT_ANSWERING: return _queueIncompleteWork.enque(work); case PARSING_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, false); case DATAPLANE_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, true); case UNKNOWN: return _queueIncompleteWork.enque(work); default: throw new BatfishException("Unknown WorkType " + work.getDetails().getWorkType()); } }
WorkQueueMgr { public synchronized boolean queueUnassignedWork(QueuedWork work) throws Exception { QueuedWork previouslyQueuedWork = getWork(work.getId()); if (previouslyQueuedWork != null) { throw new BatfishException("Duplicate work item"); } WorkDetails wDetails = work.getDetails(); cleanUpInitMetaDataIfNeeded(work.getDetails().getNetworkId(), wDetails.getSnapshotId()); if (work.getDetails().isDifferential()) { cleanUpInitMetaDataIfNeeded( work.getDetails().getNetworkId(), wDetails.getReferenceSnapshotId()); } switch (work.getDetails().getWorkType()) { case PARSING: return queueParsingWork(work); case DATAPLANING: return queueDataplaningWork(work); case INDEPENDENT_ANSWERING: return _queueIncompleteWork.enque(work); case PARSING_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, false); case DATAPLANE_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, true); case UNKNOWN: return _queueIncompleteWork.enque(work); default: throw new BatfishException("Unknown WorkType " + work.getDetails().getWorkType()); } } }
WorkQueueMgr { public synchronized boolean queueUnassignedWork(QueuedWork work) throws Exception { QueuedWork previouslyQueuedWork = getWork(work.getId()); if (previouslyQueuedWork != null) { throw new BatfishException("Duplicate work item"); } WorkDetails wDetails = work.getDetails(); cleanUpInitMetaDataIfNeeded(work.getDetails().getNetworkId(), wDetails.getSnapshotId()); if (work.getDetails().isDifferential()) { cleanUpInitMetaDataIfNeeded( work.getDetails().getNetworkId(), wDetails.getReferenceSnapshotId()); } switch (work.getDetails().getWorkType()) { case PARSING: return queueParsingWork(work); case DATAPLANING: return queueDataplaningWork(work); case INDEPENDENT_ANSWERING: return _queueIncompleteWork.enque(work); case PARSING_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, false); case DATAPLANE_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, true); case UNKNOWN: return _queueIncompleteWork.enque(work); default: throw new BatfishException("Unknown WorkType " + work.getDetails().getWorkType()); } } WorkQueueMgr(BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager);   WorkQueueMgr(Type wqType, BatfishLogger logger, SnapshotMetadataMgr snapshotMetadataManager); }
WorkQueueMgr { public synchronized boolean queueUnassignedWork(QueuedWork work) throws Exception { QueuedWork previouslyQueuedWork = getWork(work.getId()); if (previouslyQueuedWork != null) { throw new BatfishException("Duplicate work item"); } WorkDetails wDetails = work.getDetails(); cleanUpInitMetaDataIfNeeded(work.getDetails().getNetworkId(), wDetails.getSnapshotId()); if (work.getDetails().isDifferential()) { cleanUpInitMetaDataIfNeeded( work.getDetails().getNetworkId(), wDetails.getReferenceSnapshotId()); } switch (work.getDetails().getWorkType()) { case PARSING: return queueParsingWork(work); case DATAPLANING: return queueDataplaningWork(work); case INDEPENDENT_ANSWERING: return _queueIncompleteWork.enque(work); case PARSING_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, false); case DATAPLANE_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, true); case UNKNOWN: return _queueIncompleteWork.enque(work); default: throw new BatfishException("Unknown WorkType " + work.getDetails().getWorkType()); } } 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 boolean queueUnassignedWork(QueuedWork work) throws Exception { QueuedWork previouslyQueuedWork = getWork(work.getId()); if (previouslyQueuedWork != null) { throw new BatfishException("Duplicate work item"); } WorkDetails wDetails = work.getDetails(); cleanUpInitMetaDataIfNeeded(work.getDetails().getNetworkId(), wDetails.getSnapshotId()); if (work.getDetails().isDifferential()) { cleanUpInitMetaDataIfNeeded( work.getDetails().getNetworkId(), wDetails.getReferenceSnapshotId()); } switch (work.getDetails().getWorkType()) { case PARSING: return queueParsingWork(work); case DATAPLANING: return queueDataplaningWork(work); case INDEPENDENT_ANSWERING: return _queueIncompleteWork.enque(work); case PARSING_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, false); case DATAPLANE_DEPENDENT_ANSWERING: return queueDependentAnsweringWork(work, true); case UNKNOWN: return _queueIncompleteWork.enque(work); default: throw new BatfishException("Unknown WorkType " + work.getDetails().getWorkType()); } } 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);  }