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

1 files changed, 10 insertions, 5 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);
             }
         };