Handle the pipeline being aborted while finalizing an async operation. (

#1372)
PegaSysEng · Apr 30, 2019 · 79ca24c · 79ca24c
1 parent a3bff1d
commit 79ca24c
Show file tree

Hide file tree

Showing 5 changed files with 38 additions and 10 deletions.
diff --git a/...peline/src/main/java/tech/pegasys/pantheon/services/pipeline/AsyncOperationProcessor.java b/...peline/src/main/java/tech/pegasys/pantheon/services/pipeline/AsyncOperationProcessor.java
@@ -64,10 +64,9 @@ public void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> out
   }
 
   @Override
-  public void finalize(final WritePipe<O> outputPipe) {
-    while (!inProgress.isEmpty()) {
-      outputNextCompletedTask(outputPipe);
-    }
+  public boolean attemptFinalization(final WritePipe<O> outputPipe) {
+    outputNextCompletedTask(outputPipe);
+    return inProgress.isEmpty();
   }
 
   @Override

diff --git a/services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/ProcessingStage.java b/services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/ProcessingStage.java
@@ -38,7 +38,12 @@ public void run() {
     if (inputPipe.isAborted()) {
       processor.abort();
     }
-    processor.finalize(outputPipe);
+    while (!processor.attemptFinalization(outputPipe)) {
+      if (inputPipe.isAborted()) {
+        processor.abort();
+        break;
+      }
+    }
     outputPipe.close();
   }
 

diff --git a/services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/Processor.java b/services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/Processor.java
@@ -15,7 +15,9 @@
 interface Processor<I, O> {
   void processNextInput(final ReadPipe<I> inputPipe, final WritePipe<O> outputPipe);
 
-  default void finalize(final WritePipe<O> outputPipe) {}
+  default boolean attemptFinalization(final WritePipe<O> outputPipe) {
+    return true;
+  }
 
   default void abort() {}
 }
diff --git a/...ne/src/test/java/tech/pegasys/pantheon/services/pipeline/AsyncOperationProcessorTest.java b/...ne/src/test/java/tech/pegasys/pantheon/services/pipeline/AsyncOperationProcessorTest.java
@@ -94,7 +94,7 @@ public void shouldOutputRemainingInProgressTasksWhenFinalizing() {
     task2.complete("b");
 
     // Processing
-    processor.finalize(writePipe);
+    processor.attemptFinalization(writePipe);
     verify(writePipe).put("a");
     verify(writePipe).put("b");
   }
@@ -174,7 +174,7 @@ public void shouldPreserveOrderWhenRequested() {
     // And should finalize in order
     task4.complete("d");
     task3.complete("c");
-    processor.finalize(writePipe);
+    assertThat(processor.attemptFinalization(writePipe)).isTrue();
     verify(writePipe).put("c");
     verify(writePipe).put("d");
   }

diff --git a/...s/pipeline/src/test/java/tech/pegasys/pantheon/services/pipeline/ProcessingStageTest.java b/...s/pipeline/src/test/java/tech/pegasys/pantheon/services/pipeline/ProcessingStageTest.java
@@ -17,6 +17,7 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 import static tech.pegasys.pantheon.metrics.noop.NoOpMetricsSystem.NO_OP_COUNTER;
 
 import java.util.Locale;
@@ -51,6 +52,7 @@ public void setUp() {
 
   @Test
   public void shouldCallSingleStepStageForEachInput() {
+    when(singleStep.attemptFinalization(outputPipe)).thenReturn(true);
     inputPipe.put("A");
     inputPipe.put("B");
     inputPipe.put("C");
@@ -68,22 +70,42 @@ public void shouldCallSingleStepStageForEachInput() {
 
   @Test
   public void shouldFinalizeSingleStepStageAndCloseOutputPipeWhenInputCloses() {
+    when(singleStep.attemptFinalization(outputPipe)).thenReturn(true);
     inputPipe.close();
 
     stage.run();
 
-    verify(singleStep).finalize(outputPipe);
+    verify(singleStep).attemptFinalization(outputPipe);
+    verifyNoMoreInteractions(singleStep);
+    assertThat(outputPipe.isOpen()).isFalse();
+  }
+
+  @Test
+  public void shouldAbortIfPipeIsCancelledWhileAttemptingToFinalise() {
+    when(singleStep.attemptFinalization(outputPipe))
+        .thenAnswer(
+            invocation -> {
+              inputPipe.abort();
+              return false;
+            });
+    inputPipe.close();
+
+    stage.run();
+
+    verify(singleStep).attemptFinalization(outputPipe);
+    verify(singleStep).abort();
     verifyNoMoreInteractions(singleStep);
     assertThat(outputPipe.isOpen()).isFalse();
   }
 
   @Test
   public void shouldAbortProcessorIfReadPipeIsAborted() {
+    when(singleStep.attemptFinalization(outputPipe)).thenReturn(true);
     inputPipe.abort();
     stage.run();
 
     verify(singleStep).abort();
-    verify(singleStep).finalize(outputPipe);
+    verify(singleStep).attemptFinalization(outputPipe);
     assertThat(outputPipe.isOpen()).isFalse();
   }
 }