Exception not propagated by processResponse

If the topic processResponse() method throws an exception, then
the actor/operation is left in an incomplete state.

Issue-ID: POLICY-2434
Signed-off-by: Jim Hahn <jrh3@att.com>
Change-Id: I6c5d149d4046fbfb970c8dd831fc3938516d1115

2 files changed, 41 insertions, 6 deletions

diff --git a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
index ec522a405..f598d627a 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/main/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperation.java
@@ -106,7 +106,7 @@ public abstract class BidirectionalTopicOperation<Q, S> extends OperationPartial
     @Override
     protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
 
-        final Pair<String,Q> pair = makeRequest(attempt);
+        final Pair<String, Q> pair = makeRequest(attempt);
         final Q request = pair.getRight();
         outcome.setSubRequestId(pair.getLeft());
 
@@ -118,10 +118,15 @@ public abstract class BidirectionalTopicOperation<Q, S> extends OperationPartial
         // register a listener BEFORE publishing
 
         BiConsumer<String, StandardCoderObject> listener = (rawResponse, scoResponse) -> {
-            OperationOutcome latestOutcome = processResponse(outcome, rawResponse, scoResponse);
-            if (latestOutcome != null) {
-                // final response - complete the controller
-                controller.completeAsync(() -> latestOutcome, executor);
+            try {
+                OperationOutcome latestOutcome = processResponse(outcome, rawResponse, scoResponse);
+                if (latestOutcome != null) {
+                    // final response - complete the controller
+                    controller.completeAsync(() -> latestOutcome, executor);
+                }
+            } catch (RuntimeException e) {
+                logger.warn("{}: failed to process response for {}", getFullName(), params.getRequestId());
+                controller.completeExceptionally(e);
             }
         };
 
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java
index 587564a2e..48669f799 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/BidirectionalTopicOperationTest.java
@@ -168,10 +168,40 @@ public class BidirectionalTopicOperationTest {
     }
 
     /**
+     * Tests startOperationAsync() when processResponse() throws an exception.
+     */
+    @Test
+    public void testStartOperationAsyncProcException() throws Exception {
+        oper = new MyOperation() {
+            @Override
+            protected OperationOutcome processResponse(OperationOutcome outcome, String rawResponse,
+                            StandardCoderObject scoResponse) {
+                throw EXPECTED_EXCEPTION;
+            }
+        };
+
+        CompletableFuture<OperationOutcome> future = oper.startOperationAsync(1, outcome);
+        assertFalse(future.isDone());
+
+        assertEquals(SUB_REQID, outcome.getSubRequestId());
+
+        verify(forwarder).register(eq(Arrays.asList(REQ_ID)), listenerCaptor.capture());
+
+        verify(forwarder, never()).unregister(any(), any());
+
+        // provide a response
+        listenerCaptor.getValue().accept(responseText, stdResponse);
+        assertTrue(executor.runAll(MAX_REQUESTS));
+        assertTrue(future.isCompletedExceptionally());
+
+        verify(forwarder).unregister(eq(Arrays.asList(REQ_ID)), eq(listenerCaptor.getValue()));
+    }
+
+    /**
      * Tests startOperationAsync() when the publisher throws an exception.
      */
     @Test
-    public void testStartOperationAsyncException() throws Exception {
+    public void testStartOperationAsyncPubException() throws Exception {
         // indicate that nothing was published
         when(handler.send(any())).thenReturn(false);