Interrupt matchStage to induce prepare shutdown

The only signal to a waiting match that will halt its current listen
loop for a valid unique operation is an interrupt.

src/main/java/build/buildfarm/worker/Pipeline.java

@@ -61,9 +61,12 @@ public void close() throws InterruptedException {
 
   /** Inform MatchStage to stop matching or picking up new Operations from queue. */
   public void stopMatchingOperations() {
-    for (PipelineStage stage : stageClosePriorities.keySet()) {
+    for (Map.Entry<PipelineStage, Thread> entry : stageThreads.entrySet()) {
+      PipelineStage stage = entry.getKey();
       if (stage instanceof MatchStage) {
-        ((MatchStage) stage).prepareForGracefulShutdown();
+        MatchStage matchStage = (MatchStage) stage;
+        matchStage.prepareForGracefulShutdown();
+        entry.getValue().interrupt();
         return;
       }
     }
@@ -134,15 +137,17 @@ private void join(boolean closeStage) throws InterruptedException {
             }
           }
           if (stageToClose != null && !stageToClose.isClosed()) {
-            log.log(Level.FINER, "Closing stage at priority " + maxPriority);
+            log.log(Level.FINER, "Closing stage " + stageToClose + " at priority " + maxPriority);
             stageToClose.close();
           }
         }
+        boolean longStageWait = !closeStage;
         for (Map.Entry<PipelineStage, Thread> stageThread : stageThreads.entrySet()) {
           PipelineStage stage = stageThread.getKey();
           Thread thread = stageThread.getValue();
           try {
-            thread.join(closeStage ? 1 : 1000);
+            // 0 is wait forever, no instant wait
+            thread.join(longStageWait ? 1000 : 1);
           } catch (InterruptedException e) {
             if (!closeStage) {
               synchronized (this) {
@@ -172,8 +177,8 @@ private void join(boolean closeStage) throws InterruptedException {
                     + stageClosePriorities.get(stage));
             thread.interrupt();
           }
+          longStageWait = false;
         }
-        closeStage = false;
         for (PipelineStage stage : inactiveStages) {
           synchronized (this) {
             stageThreads.remove(stage);