Clean up and enhancement of Actor re-design

Added junits for the remaining code.

Enhancements to facilitate implementation of Operators:
- Added allOf(), anyOf() facilities
- Added AsyncResponseHandler for handling asynchronous I/O via the
  HttpClient
- Added logRestRequest() and logRestResponse() for logging REST
  requests and responses
- Added HttpActor and HttpOperator, which can be used as superclasses
- Added doTask()
- Lifted data from the event into ControlLoopEventContext

Updates per previous review comments:
- Changed logException() to runFunction().
- Removed the aaiCqResponse field.
- Lifted fields from Policy into ControlLoopOperationParams, eliminating
  the need to include Policy in the class.

OperatorPartial depends on the string values in the ControlLoopOperation
being set to one of the string values of PolicyResult.  Instead of
passing ControlLoopOperation around, the operators should pass around
an object that uses PolicyResult directly, rather than depending on
the string values being set correctly.  Created OperationOutcome for
this purpose.
Stop pipeline when the controller completes.
Use whenComplete() where appropriate.
startOperationAsync() should not block.  Modified it to launch the task
in the background via its own thread.
Extracted CallbackManager into its own file.
Replaced actor setOperators() with addOperator()
Renamed add() to wrap(), and modified it to remove the future when it
completes.
Fixed the signature on delayedRemove() and delayedComplete().
Replaced xxxAsync() calls with just xxx() calls, where appropriate to
avoid the extra overhead of submitting it to a work queue.
Renamed handleFailure() to handlePreprocessorFailure().
Updates per WIP review comments

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

14 files changed, 1627 insertions, 326 deletions

diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java
new file mode 100644
index 000000000..31c6d2077
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/AsyncResponseHandlerTest.java
@@ -0,0 +1,172 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.policy.PolicyResult;
+
+public class AsyncResponseHandlerTest {
+
+    private static final String ACTOR = "my-actor";
+    private static final String OPERATION = "my-operation";
+    private static final UUID REQ_ID = UUID.randomUUID();
+    private static final String TEXT = "some text";
+
+    private VirtualControlLoopEvent event;
+    private ControlLoopEventContext context;
+    private ControlLoopOperationParams params;
+    private OperationOutcome outcome;
+    private MyHandler handler;
+
+    /**
+     * Initializes all fields, including {@link #handler}.
+     */
+    @Before
+    public void setUp() {
+        event = new VirtualControlLoopEvent();
+        event.setRequestId(REQ_ID);
+
+        context = new ControlLoopEventContext(event);
+        params = ControlLoopOperationParams.builder().actor(ACTOR).operation(OPERATION).context(context).build();
+        outcome = params.makeOutcome();
+
+        handler = new MyHandler(params, outcome);
+    }
+
+    @Test
+    public void testAsyncResponseHandler_testGetParams_testGetOutcome() {
+        assertSame(params, handler.getParams());
+        assertSame(outcome, handler.getOutcome());
+    }
+
+    @Test
+    public void testHandle() {
+        CompletableFuture<String> future = new CompletableFuture<>();
+        handler.handle(future).complete(outcome);
+
+        assertTrue(future.isCancelled());
+    }
+
+    @Test
+    public void testCompleted() throws Exception {
+        CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+        handler.completed(TEXT);
+        assertTrue(result.isDone());
+        assertSame(outcome, result.get());
+        assertEquals(PolicyResult.FAILURE_RETRIES, outcome.getResult());
+        assertEquals(TEXT, outcome.getMessage());
+    }
+
+    /**
+     * Tests completed() when doCompleted() throws an exception.
+     */
+    @Test
+    public void testCompletedException() throws Exception {
+        IllegalStateException except = new IllegalStateException();
+
+        outcome = params.makeOutcome();
+        handler = new MyHandler(params, outcome) {
+            @Override
+            protected OperationOutcome doComplete(String rawResponse) {
+                throw except;
+            }
+        };
+
+        CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+        handler.completed(TEXT);
+        assertTrue(result.isCompletedExceptionally());
+
+        AtomicReference<Throwable> thrown = new AtomicReference<>();
+        result.whenComplete((unused, thrown2) -> thrown.set(thrown2));
+
+        assertSame(except, thrown.get());
+    }
+
+    @Test
+    public void testFailed() throws Exception {
+        IllegalStateException except = new IllegalStateException();
+
+        CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+        handler.failed(except);
+
+        assertTrue(result.isDone());
+        assertSame(outcome, result.get());
+        assertEquals(PolicyResult.FAILURE_GUARD, outcome.getResult());
+    }
+
+    /**
+     * Tests failed() when doFailed() throws an exception.
+     */
+    @Test
+    public void testFailedException() throws Exception {
+        IllegalStateException except = new IllegalStateException();
+
+        outcome = params.makeOutcome();
+        handler = new MyHandler(params, outcome) {
+            @Override
+            protected OperationOutcome doFailed(Throwable thrown) {
+                throw except;
+            }
+        };
+
+        CompletableFuture<OperationOutcome> result = handler.handle(new CompletableFuture<>());
+        handler.failed(except);
+        assertTrue(result.isCompletedExceptionally());
+
+        AtomicReference<Throwable> thrown = new AtomicReference<>();
+        result.whenComplete((unused, thrown2) -> thrown.set(thrown2));
+
+        assertSame(except, thrown.get());
+    }
+
+    private class MyHandler extends AsyncResponseHandler<String> {
+
+        public MyHandler(ControlLoopOperationParams params, OperationOutcome outcome) {
+            super(params, outcome);
+        }
+
+        @Override
+        protected OperationOutcome doComplete(String rawResponse) {
+            OperationOutcome outcome = getOutcome();
+            outcome.setResult(PolicyResult.FAILURE_RETRIES);
+            outcome.setMessage(rawResponse);
+            return outcome;
+        }
+
+        @Override
+        protected OperationOutcome doFailed(Throwable thrown) {
+            OperationOutcome outcome = getOutcome();
+            outcome.setResult(PolicyResult.FAILURE_GUARD);
+            return outcome;
+        }
+    }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManagerTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManagerTest.java
new file mode 100644
index 000000000..44606cb14
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/CallbackManagerTest.java
@@ -0,0 +1,89 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.time.Instant;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CallbackManagerTest {
+
+    private CallbackManager mgr;
+
+    @Before
+    public void setUp() {
+        mgr = new CallbackManager();
+    }
+
+    @Test
+    public void testCanStart_testGetStartTime() {
+        // null until canXxx() is called
+        assertNull(mgr.getStartTime());
+
+        assertTrue(mgr.canStart());
+
+        Instant time = mgr.getStartTime();
+        assertNotNull(time);
+        assertNull(mgr.getEndTime());
+
+        // false for now on
+        assertFalse(mgr.canStart());
+        assertFalse(mgr.canStart());
+
+        assertEquals(time, mgr.getStartTime());
+    }
+
+    @Test
+    public void testCanEnd_testGetEndTime() {
+        // null until canXxx() is called
+        assertNull(mgr.getEndTime());
+        assertNull(mgr.getEndTime());
+
+        assertTrue(mgr.canEnd());
+
+        Instant time = mgr.getEndTime();
+        assertNotNull(time);
+        assertNull(mgr.getStartTime());
+
+        // false for now on
+        assertFalse(mgr.canEnd());
+        assertFalse(mgr.canEnd());
+
+        assertEquals(time, mgr.getEndTime());
+    }
+
+    @Test
+    public void testRun() {
+        mgr.run();
+
+        assertNotNull(mgr.getStartTime());
+        assertNotNull(mgr.getEndTime());
+
+        assertFalse(mgr.canStart());
+        assertFalse(mgr.canEnd());
+    }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcomeTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcomeTest.java
new file mode 100644
index 000000000..4e9728336
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/OperationOutcomeTest.java
@@ -0,0 +1,137 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.time.Instant;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.controlloop.ControlLoopOperation;
+import org.onap.policy.controlloop.policy.PolicyResult;
+
+public class OperationOutcomeTest {
+    private static final String ACTOR = "my-actor";
+    private static final String OPERATION = "my-operation";
+    private static final String TARGET = "my-target";
+    private static final Instant START = Instant.ofEpochMilli(10);
+    private static final Instant END = Instant.ofEpochMilli(20);
+    private static final String SUB_REQ_ID = "my-sub-request-id";
+    private static final PolicyResult RESULT = PolicyResult.FAILURE_GUARD;
+    private static final String MESSAGE = "my-message";
+
+    private OperationOutcome outcome;
+
+    @Before
+    public void setUp() {
+        outcome = new OperationOutcome();
+    }
+
+    @Test
+    public void testOperationOutcomeOperationOutcome() {
+        setAll();
+
+        OperationOutcome outcome2 = new OperationOutcome(outcome);
+
+        assertEquals(ACTOR, outcome2.getActor());
+        assertEquals(OPERATION, outcome2.getOperation());
+        assertEquals(TARGET, outcome2.getTarget());
+        assertEquals(START, outcome2.getStart());
+        assertEquals(END, outcome2.getEnd());
+        assertEquals(SUB_REQ_ID, outcome2.getSubRequestId());
+        assertEquals(RESULT, outcome2.getResult());
+        assertEquals(MESSAGE, outcome2.getMessage());
+    }
+
+    @Test
+    public void testToControlLoopOperation() {
+        setAll();
+
+        ControlLoopOperation outcome2 = outcome.toControlLoopOperation();
+
+        assertEquals(ACTOR, outcome2.getActor());
+        assertEquals(OPERATION, outcome2.getOperation());
+        assertEquals(TARGET, outcome2.getTarget());
+        assertEquals(START, outcome2.getStart());
+        assertEquals(END, outcome2.getEnd());
+        assertEquals(SUB_REQ_ID, outcome2.getSubRequestId());
+        assertEquals(RESULT.toString(), outcome2.getOutcome());
+        assertEquals(MESSAGE, outcome2.getMessage());
+    }
+
+    /**
+     * Tests both isFor() methods, as one invokes the other.
+     */
+    @Test
+    public void testIsFor() {
+        setAll();
+
+        // null case
+        assertFalse(OperationOutcome.isFor(null, ACTOR, OPERATION));
+
+        // actor mismatch
+        assertFalse(OperationOutcome.isFor(outcome, TARGET, OPERATION));
+
+        // operation mismatch
+        assertFalse(OperationOutcome.isFor(outcome, ACTOR, TARGET));
+
+        // null actor in outcome
+        outcome.setActor(null);
+        assertFalse(OperationOutcome.isFor(outcome, ACTOR, OPERATION));
+        outcome.setActor(ACTOR);
+
+        // null operation in outcome
+        outcome.setOperation(null);
+        assertFalse(OperationOutcome.isFor(outcome, ACTOR, OPERATION));
+        outcome.setOperation(OPERATION);
+
+        // null actor argument
+        assertThatThrownBy(() -> outcome.isFor(null, OPERATION));
+
+        // null operation argument
+        assertThatThrownBy(() -> outcome.isFor(ACTOR, null));
+
+        // true case
+        assertTrue(OperationOutcome.isFor(outcome, ACTOR, OPERATION));
+    }
+
+    @Test
+    public void testSetResult() {
+        outcome.setResult(PolicyResult.FAILURE_EXCEPTION);
+        assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
+
+        assertThatThrownBy(() -> outcome.setResult(null));
+    }
+
+    private void setAll() {
+        outcome.setActor(ACTOR);
+        outcome.setEnd(END);
+        outcome.setMessage(MESSAGE);
+        outcome.setOperation(OPERATION);
+        outcome.setResult(RESULT);
+        outcome.setStart(START);
+        outcome.setSubRequestId(SUB_REQ_ID);
+        outcome.setTarget(TARGET);
+    }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java
index c652e8374..4a3f321cf 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/UtilTest.java
@@ -27,15 +27,56 @@ import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+import ch.qos.logback.classic.Logger;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.Builder;
 import lombok.Data;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.common.utils.test.log.logback.ExtractAppender;
+import org.slf4j.LoggerFactory;
 
 public class UtilTest {
+    private static final String MY_REQUEST = "my-request";
+    private static final String URL = "my-url";
+    private static final String OUT_URL = "OUT|REST|my-url";
+    private static final String IN_URL = "IN|REST|my-url";
+    protected static final String EXPECTED_EXCEPTION = "expected exception";
+
+    /**
+     * Used to attach an appender to the class' logger.
+     */
+    private static final Logger logger = (Logger) LoggerFactory.getLogger(Util.class);
+    private static final ExtractAppender appender = new ExtractAppender();
+
+    /**
+     * Initializes statics.
+     */
+    @BeforeClass
+    public static void setUpBeforeClass() {
+        appender.setContext(logger.getLoggerContext());
+        appender.start();
+
+        logger.addAppender(appender);
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() {
+        appender.stop();
+    }
+
+    @Before
+    public void setUp() {
+        appender.clearExtractions();
+    }
 
     @Test
     public void testIdent() {
@@ -48,11 +89,88 @@ public class UtilTest {
     }
 
     @Test
-    public void testLogException() {
+    public void testLogRestRequest() throws CoderException {
+        // log structured data
+        appender.clearExtractions();
+        Util.logRestRequest(URL, new Abc(10, null, null));
+        List<String> output = appender.getExtracted();
+        assertEquals(1, output.size());
+
+        assertThat(output.get(0)).contains(OUT_URL).contains("{\n  \"intValue\": 10\n}");
+
+        // log a plain string
+        appender.clearExtractions();
+        Util.logRestRequest(URL, MY_REQUEST);
+        output = appender.getExtracted();
+        assertEquals(1, output.size());
+
+        assertThat(output.get(0)).contains(OUT_URL).contains(MY_REQUEST);
+
+        // exception from coder
+        StandardCoder coder = new StandardCoder() {
+            @Override
+            public String encode(Object object, boolean pretty) throws CoderException {
+                throw new CoderException(EXPECTED_EXCEPTION);
+            }
+        };
+
+        appender.clearExtractions();
+        Util.logRestRequest(coder, URL, new Abc(11, null, null));
+        output = appender.getExtracted();
+        assertEquals(2, output.size());
+        assertThat(output.get(0)).contains("cannot pretty-print request");
+        assertThat(output.get(1)).contains(OUT_URL);
+    }
+
+    @Test
+    public void testLogRestResponse() throws CoderException {
+        // log structured data
+        appender.clearExtractions();
+        Util.logRestResponse(URL, new Abc(10, null, null));
+        List<String> output = appender.getExtracted();
+        assertEquals(1, output.size());
+
+        assertThat(output.get(0)).contains(IN_URL).contains("{\n  \"intValue\": 10\n}");
+
+        // log null response
+        appender.clearExtractions();
+        Util.logRestResponse(URL, null);
+        output = appender.getExtracted();
+        assertEquals(1, output.size());
+
+        assertThat(output.get(0)).contains(IN_URL).contains("null");
+
+        // log a plain string
+        appender.clearExtractions();
+        Util.logRestResponse(URL, MY_REQUEST);
+        output = appender.getExtracted();
+        assertEquals(1, output.size());
+
+        assertThat(output.get(0)).contains(IN_URL).contains(MY_REQUEST);
+
+        // exception from coder
+        StandardCoder coder = new StandardCoder() {
+            @Override
+            public String encode(Object object, boolean pretty) throws CoderException {
+                throw new CoderException(EXPECTED_EXCEPTION);
+            }
+        };
+
+        appender.clearExtractions();
+        Util.logRestResponse(coder, URL, new Abc(11, null, null));
+        output = appender.getExtracted();
+        assertEquals(2, output.size());
+        assertThat(output.get(0)).contains("cannot pretty-print response");
+        assertThat(output.get(1)).contains(IN_URL);
+    }
+
+    @Test
+    public void testRunFunction() {
         // no exception, no log
         AtomicInteger count = new AtomicInteger();
-        Util.logException(() -> count.incrementAndGet(), "no error");
+        Util.runFunction(() -> count.incrementAndGet(), "no error");
         assertEquals(1, count.get());
+        assertEquals(0, appender.getExtracted().size());
 
         // with an exception
         Runnable runnable = () -> {
@@ -60,8 +178,17 @@ public class UtilTest {
             throw new IllegalStateException("expected exception");
         };
 
-        Util.logException(runnable, "error with no args");
-        Util.logException(runnable, "error {} {} arg(s)", "with", 1);
+        appender.clearExtractions();
+        Util.runFunction(runnable, "error with no args");
+        List<String> output = appender.getExtracted();
+        assertEquals(1, output.size());
+        assertThat(output.get(0)).contains("error with no args");
+
+        appender.clearExtractions();
+        Util.runFunction(runnable, "error {} {} arg(s)", "with", 2);
+        output = appender.getExtracted();
+        assertEquals(1, output.size());
+        assertThat(output.get(0)).contains("error with 2 arg(s)");
     }
 
     @Test
@@ -123,4 +250,14 @@ public class UtilTest {
         private int intValue;
         private String strValue;
     }
+
+    // throws an exception when getXxx() is used
+    public static class DataWithException {
+        @SuppressWarnings("unused")
+        private int intValue;
+
+        public int getIntValue() {
+            throw new IllegalStateException();
+        }
+    }
 }
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java
index fcc3fb12e..0d917ad3e 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/controlloop/ControlLoopEventContextTest.java
@@ -20,28 +20,55 @@
 
 package org.onap.policy.controlloop.actorserviceprovider.controlloop;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
 
+import java.util.Map;
+import java.util.UUID;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.controlloop.VirtualControlLoopEvent;
 
 public class ControlLoopEventContextTest {
+    private static final UUID REQ_ID = UUID.randomUUID();
 
+    private Map<String, String> enrichment;
     private VirtualControlLoopEvent event;
     private ControlLoopEventContext context;
 
+    /**
+     * Initializes data, including {@link #context}.
+     */
     @Before
     public void setUp() {
+        enrichment = Map.of("abc", "one", "def", "two");
+
         event = new VirtualControlLoopEvent();
+        event.setRequestId(REQ_ID);
+        event.setAai(enrichment);
+
         context = new ControlLoopEventContext(event);
     }
 
     @Test
     public void testControlLoopEventContext() {
         assertSame(event, context.getEvent());
+        assertSame(REQ_ID, context.getRequestId());
+        assertEquals(enrichment, context.getEnrichment());
+
+        // null event
+        assertThatThrownBy(() -> new ControlLoopEventContext(null));
+
+        // no request id, no enrichment data
+        event.setRequestId(null);
+        event.setAai(null);
+        context = new ControlLoopEventContext(event);
+        assertSame(event, context.getEvent());
+        assertNotNull(context.getRequestId());
+        assertEquals(Map.of(), context.getEnrichment());
     }
 
     @Test
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java
index 7e0c35a3f..a209fb0d8 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/ActorImplTest.java
@@ -34,7 +34,6 @@ import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-import java.util.Collections;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.TreeMap;
@@ -192,10 +191,10 @@ public class ActorImplTest {
     }
 
     @Test
-    public void testSetOperators() {
-        // cannot set operators if already configured
+    public void testAddOperator() {
+        // cannot add operators if already configured
         actor.configure(params);
-        assertThatIllegalStateException().isThrownBy(() -> actor.setOperators(Collections.emptyList()));
+        assertThatIllegalStateException().isThrownBy(() -> actor.addOperator(oper1));
 
         /*
          * make an actor where operators two and four have names that are duplicates of
@@ -367,7 +366,13 @@ public class ActorImplTest {
      * @return a new actor
      */
     private ActorImpl makeActor(Operator... operators) {
-        return new ActorImpl(ACTOR_NAME, operators);
+        ActorImpl actor = new ActorImpl(ACTOR_NAME);
+
+        for (Operator oper : operators) {
+            actor.addOperator(oper);
+        }
+
+        return actor;
     }
 
     private static class MyOper extends OperatorPartial implements Operator {
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java
new file mode 100644
index 000000000..2da789989
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpActorTest.java
@@ -0,0 +1,81 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider.impl;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.function.Function;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpActorParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
+
+public class HttpActorTest {
+
+    private static final String ACTOR = "my-actor";
+    private static final String UNKNOWN = "unknown";
+    private static final String CLIENT = "my-client";
+    private static final long TIMEOUT = 10L;
+
+    private HttpActor actor;
+
+    @Before
+    public void setUp() {
+        actor = new HttpActor(ACTOR);
+    }
+
+    @Test
+    public void testMakeOperatorParameters() {
+        HttpActorParams params = new HttpActorParams();
+        params.setClientName(CLIENT);
+        params.setTimeoutSec(TIMEOUT);
+        params.setPath(Map.of("operA", "urlA", "operB", "urlB"));
+
+        final HttpActor prov = new HttpActor(ACTOR);
+        Function<String, Map<String, Object>> maker =
+                        prov.makeOperatorParameters(Util.translateToMap(prov.getName(), params));
+
+        assertNull(maker.apply(UNKNOWN));
+
+        // use a TreeMap to ensure the properties are sorted
+        assertEquals("{clientName=my-client, path=urlA, timeoutSec=10}",
+                        new TreeMap<>(maker.apply("operA")).toString());
+
+        assertEquals("{clientName=my-client, path=urlB, timeoutSec=10}",
+                        new TreeMap<>(maker.apply("operB")).toString());
+
+        // with invalid actor parameters
+        params.setClientName(null);
+        assertThatThrownBy(() -> prov.makeOperatorParameters(Util.translateToMap(prov.getName(), params)))
+                        .isInstanceOf(ParameterValidationRuntimeException.class);
+    }
+
+    @Test
+    public void testHttpActor() {
+        assertEquals(ACTOR, actor.getName());
+        assertEquals(ACTOR, actor.getFullName());
+    }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java
new file mode 100644
index 000000000..c006cf333
--- /dev/null
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/HttpOperatorTest.java
@@ -0,0 +1,104 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actorserviceprovider.impl;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Map;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.policy.common.endpoints.http.client.HttpClient;
+import org.onap.policy.common.endpoints.http.client.HttpClientFactory;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.HttpParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
+
+public class HttpOperatorTest {
+
+    private static final String ACTOR = "my-actor";
+    private static final String OPERATION = "my-name";
+    private static final String CLIENT = "my-client";
+    private static final String PATH = "my-path";
+    private static final long TIMEOUT = 100;
+
+    @Mock
+    private HttpClient client;
+
+    private HttpOperator oper;
+
+    /**
+     * Initializes fields, including {@link #oper}.
+     */
+    @Before
+    public void setUp() {
+        MockitoAnnotations.initMocks(this);
+
+        oper = new HttpOperator(ACTOR, OPERATION);
+    }
+
+    @Test
+    public void testDoConfigureMapOfStringObject_testGetClient_testGetPath_testGetTimeoutSec() {
+        assertNull(oper.getClient());
+        assertNull(oper.getPath());
+        assertEquals(0L, oper.getTimeoutSec());
+
+        oper = new HttpOperator(ACTOR, OPERATION) {
+            @Override
+            protected HttpClientFactory getClientFactory() {
+                HttpClientFactory factory = mock(HttpClientFactory.class);
+                when(factory.get(CLIENT)).thenReturn(client);
+                return factory;
+            }
+        };
+
+        HttpParams params = HttpParams.builder().clientName(CLIENT).path(PATH).timeoutSec(TIMEOUT).build();
+        Map<String, Object> paramMap = Util.translateToMap(OPERATION, params);
+        oper.configure(paramMap);
+
+        assertSame(client, oper.getClient());
+        assertEquals(PATH, oper.getPath());
+        assertEquals(TIMEOUT, oper.getTimeoutSec());
+
+        // test invalid parameters
+        paramMap.remove("path");
+        assertThatThrownBy(() -> oper.configure(paramMap)).isInstanceOf(ParameterValidationRuntimeException.class);
+    }
+
+    @Test
+    public void testHttpOperator() {
+        assertEquals(ACTOR, oper.getActorName());
+        assertEquals(OPERATION, oper.getName());
+        assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
+    }
+
+    @Test
+    public void testGetClient() {
+        assertNotNull(oper.getClientFactory());
+    }
+}
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java
index 864ac829a..21bc656f2 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperatorPartialTest.java
@@ -29,6 +29,7 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.time.Instant;
@@ -36,17 +37,20 @@ import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Queue;
 import java.util.TreeMap;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
@@ -58,9 +62,10 @@ import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.controlloop.ControlLoopOperation;
 import org.onap.policy.controlloop.VirtualControlLoopEvent;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
 import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
-import org.onap.policy.controlloop.policy.Policy;
+import org.onap.policy.controlloop.actorserviceprovider.pipeline.PipelineControllerFuture;
 import org.onap.policy.controlloop.policy.PolicyResult;
 
 public class OperatorPartialTest {
@@ -75,14 +80,10 @@ public class OperatorPartialTest {
     private static final List<PolicyResult> FAILURE_RESULTS = Arrays.asList(PolicyResult.values()).stream()
                     .filter(result -> result != PolicyResult.SUCCESS).collect(Collectors.toList());
 
-    private static final List<String> FAILURE_STRINGS =
-                    FAILURE_RESULTS.stream().map(Object::toString).collect(Collectors.toList());
-
     private VirtualControlLoopEvent event;
     private Map<String, Object> config;
     private ControlLoopEventContext context;
     private MyExec executor;
-    private Policy policy;
     private ControlLoopOperationParams params;
 
     private MyOper oper;
@@ -92,8 +93,8 @@ public class OperatorPartialTest {
 
     private Instant tstart;
 
-    private ControlLoopOperation opstart;
-    private ControlLoopOperation opend;
+    private OperationOutcome opstart;
+    private OperationOutcome opend;
 
     /**
      * Initializes the fields, including {@link #oper}.
@@ -107,13 +108,9 @@ public class OperatorPartialTest {
         context = new ControlLoopEventContext(event);
         executor = new MyExec();
 
-        policy = new Policy();
-        policy.setActor(ACTOR);
-        policy.setRecipe(OPERATOR);
-        policy.setTimeout(TIMEOUT);
-
         params = ControlLoopOperationParams.builder().completeCallback(this::completer).context(context)
-                        .executor(executor).policy(policy).startCallback(this::starter).target(TARGET).build();
+                        .executor(executor).actor(ACTOR).operation(OPERATOR).timeoutSec(TIMEOUT)
+                        .startCallback(this::starter).targetEntity(TARGET).build();
 
         oper = new MyOper();
         oper.configure(new TreeMap<>());
@@ -133,6 +130,31 @@ public class OperatorPartialTest {
     }
 
     @Test
+    public void testGetBlockingExecutor() throws InterruptedException {
+        CountDownLatch latch = new CountDownLatch(1);
+
+        /*
+         * Use an operator that doesn't override getBlockingExecutor().
+         */
+        OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATOR) {};
+        oper2.getBlockingExecutor().execute(() -> latch.countDown());
+
+        assertTrue(latch.await(5, TimeUnit.SECONDS));
+    }
+
+    @Test
+    public void testDoConfigure() {
+        oper = spy(new MyOper());
+
+        oper.configure(config);
+        verify(oper).configure(config);
+
+        // repeat - SHOULD be run again
+        oper.configure(config);
+        verify(oper, times(2)).configure(config);
+    }
+
+    @Test
     public void testDoStart() {
         oper = spy(new MyOper());
 
@@ -181,7 +203,7 @@ public class OperatorPartialTest {
     }
 
     @Test
-    public void testStartOperation_testVerifyRunning() {
+    public void testStartOperation() {
         verifyRun("testStartOperation", 1, 1, PolicyResult.SUCCESS);
     }
 
@@ -201,17 +223,13 @@ public class OperatorPartialTest {
      * Tests startOperation() when the operation has a preprocessor.
      */
     @Test
-    public void testStartOperationWithPreprocessor_testStartPreprocessor() {
+    public void testStartOperationWithPreprocessor() {
         AtomicInteger count = new AtomicInteger();
 
-        // @formatter:off
-        Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> preproc =
-            oper -> CompletableFuture.supplyAsync(() -> {
-                count.incrementAndGet();
-                oper.setOutcome(PolicyResult.SUCCESS.toString());
-                return oper;
-            }, executor);
-        // @formatter:on
+        CompletableFuture<OperationOutcome> preproc = CompletableFuture.supplyAsync(() -> {
+            count.incrementAndGet();
+            return makeSuccess();
+        }, executor);
 
         oper.setPreProcessor(preproc);
 
@@ -233,7 +251,7 @@ public class OperatorPartialTest {
 
         assertNotNull(opstart);
         assertNotNull(opend);
-        assertEquals(PolicyResult.SUCCESS.toString(), opend.getOutcome());
+        assertEquals(PolicyResult.SUCCESS, opend.getResult());
 
         assertEquals(MAX_PARALLEL_REQUESTS, numStart);
         assertEquals(MAX_PARALLEL_REQUESTS, oper.getCount());
@@ -245,11 +263,7 @@ public class OperatorPartialTest {
      */
     @Test
     public void testStartPreprocessorFailure() {
-        // arrange for the preprocessor to return a failure
-        oper.setPreProcessor(oper -> {
-            oper.setOutcome(PolicyResult.FAILURE_GUARD.toString());
-            return CompletableFuture.completedFuture(oper);
-        });
+        oper.setPreProcessor(CompletableFuture.completedFuture(makeFailure()));
 
         verifyRun("testStartPreprocessorFailure", 1, 0, PolicyResult.FAILURE_GUARD);
     }
@@ -260,9 +274,7 @@ public class OperatorPartialTest {
     @Test
     public void testStartPreprocessorException() {
         // arrange for the preprocessor to throw an exception
-        oper.setPreProcessor(oper -> {
-            throw new IllegalStateException(EXPECTED_EXCEPTION);
-        });
+        oper.setPreProcessor(CompletableFuture.failedFuture(new IllegalStateException(EXPECTED_EXCEPTION)));
 
         verifyRun("testStartPreprocessorException", 1, 0, PolicyResult.FAILURE_GUARD);
     }
@@ -273,10 +285,7 @@ public class OperatorPartialTest {
     @Test
     public void testStartPreprocessorNotRunning() {
         // arrange for the preprocessor to return success, which will be ignored
-        oper.setPreProcessor(oper -> {
-            oper.setOutcome(PolicyResult.SUCCESS.toString());
-            return CompletableFuture.completedFuture(oper);
-        });
+        oper.setPreProcessor(CompletableFuture.completedFuture(makeSuccess()));
 
         oper.startOperation(params).cancel(false);
         assertTrue(executor.runAll());
@@ -296,8 +305,7 @@ public class OperatorPartialTest {
     public void testStartPreprocessorBuilderException() {
         oper = new MyOper() {
             @Override
-            protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doPreprocessorAsFuture(
-                            ControlLoopOperationParams params) {
+            protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
                 throw new IllegalStateException(EXPECTED_EXCEPTION);
             }
         };
@@ -312,51 +320,27 @@ public class OperatorPartialTest {
     }
 
     @Test
-    public void testDoPreprocessorAsFuture() {
-        assertNull(oper.doPreprocessorAsFuture(params));
+    public void testStartPreprocessorAsync() {
+        assertNull(oper.startPreprocessorAsync(params));
     }
 
     @Test
-    public void testStartOperationOnly_testDoOperationAsFuture() {
+    public void testStartOperationAsync() {
         oper.startOperation(params);
         assertTrue(executor.runAll());
 
         assertEquals(1, oper.getCount());
     }
 
-    /**
-     * Tests startOperationOnce() when
-     * {@link OperatorPartial#doOperationAsFuture(ControlLoopOperationParams)} throws an
-     * exception.
-     */
-    @Test
-    public void testStartOperationOnceBuilderException() {
-        oper = new MyOper() {
-            @Override
-            protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doOperationAsFuture(
-                            ControlLoopOperationParams params, int attempt) {
-                throw new IllegalStateException(EXPECTED_EXCEPTION);
-            }
-        };
-
-        oper.configure(new TreeMap<>());
-        oper.start();
-
-        assertThatIllegalStateException().isThrownBy(() -> oper.startOperation(params));
-
-        // should be nothing in the queue
-        assertEquals(0, executor.getQueueLength());
-    }
-
     @Test
     public void testIsSuccess() {
-        ControlLoopOperation outcome = new ControlLoopOperation();
+        OperationOutcome outcome = new OperationOutcome();
 
-        outcome.setOutcome(PolicyResult.SUCCESS.toString());
+        outcome.setResult(PolicyResult.SUCCESS);
         assertTrue(oper.isSuccess(outcome));
 
-        for (String failure : FAILURE_STRINGS) {
-            outcome.setOutcome(failure);
+        for (PolicyResult failure : FAILURE_RESULTS) {
+            outcome.setResult(failure);
             assertFalse("testIsSuccess-" + failure, oper.isSuccess(outcome));
         }
     }
@@ -365,17 +349,17 @@ public class OperatorPartialTest {
     public void testIsActorFailed() {
         assertFalse(oper.isActorFailed(null));
 
-        ControlLoopOperation outcome = params.makeOutcome();
+        OperationOutcome outcome = params.makeOutcome();
 
         // incorrect outcome
-        outcome.setOutcome(PolicyResult.SUCCESS.toString());
+        outcome.setResult(PolicyResult.SUCCESS);
         assertFalse(oper.isActorFailed(outcome));
 
-        outcome.setOutcome(PolicyResult.FAILURE_RETRIES.toString());
+        outcome.setResult(PolicyResult.FAILURE_RETRIES);
         assertFalse(oper.isActorFailed(outcome));
 
         // correct outcome
-        outcome.setOutcome(PolicyResult.FAILURE.toString());
+        outcome.setResult(PolicyResult.FAILURE);
 
         // incorrect actor
         outcome.setActor(TARGET);
@@ -400,7 +384,12 @@ public class OperatorPartialTest {
         /*
          * Use an operator that doesn't override doOperation().
          */
-        OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATOR) {};
+        OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATOR) {
+            @Override
+            protected Executor getBlockingExecutor() {
+                return executor;
+            }
+        };
 
         oper2.configure(new TreeMap<>());
         oper2.start();
@@ -409,7 +398,7 @@ public class OperatorPartialTest {
         assertTrue(executor.runAll());
 
         assertNotNull(opend);
-        assertEquals(PolicyResult.FAILURE_EXCEPTION.toString(), opend.getOutcome());
+        assertEquals(PolicyResult.FAILURE_EXCEPTION, opend.getResult());
     }
 
     @Test
@@ -421,36 +410,34 @@ public class OperatorPartialTest {
         // trigger timeout very quickly
         oper = new MyOper() {
             @Override
-            protected long getTimeOutMillis(Policy policy) {
+            protected long getTimeOutMillis(Integer timeoutSec) {
                 return 1;
             }
 
             @Override
-            protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doOperationAsFuture(
-                            ControlLoopOperationParams params, int attempt) {
-
-                return outcome -> {
-                    ControlLoopOperation outcome2 = params.makeOutcome();
-                    outcome2.setOutcome(PolicyResult.SUCCESS.toString());
-
-                    /*
-                     * Create an incomplete future that will timeout after the operation's
-                     * timeout. If it fires before the other timer, then it will return a
-                     * SUCCESS outcome.
-                     */
-                    CompletableFuture<ControlLoopOperation> future = new CompletableFuture<>();
-                    future = future.orTimeout(1, TimeUnit.SECONDS).handleAsync((unused1, unused2) -> outcome,
-                                    params.getExecutor());
-
-                    return future;
-                };
+            protected CompletableFuture<OperationOutcome> startOperationAsync(ControlLoopOperationParams params,
+                            int attempt, OperationOutcome outcome) {
+
+                OperationOutcome outcome2 = params.makeOutcome();
+                outcome2.setResult(PolicyResult.SUCCESS);
+
+                /*
+                 * Create an incomplete future that will timeout after the operation's
+                 * timeout. If it fires before the other timer, then it will return a
+                 * SUCCESS outcome.
+                 */
+                CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
+                future = future.orTimeout(1, TimeUnit.SECONDS).handleAsync((unused1, unused2) -> outcome,
+                                params.getExecutor());
+
+                return future;
             }
         };
 
         oper.configure(new TreeMap<>());
         oper.start();
 
-        assertEquals(PolicyResult.FAILURE_TIMEOUT.toString(), oper.startOperation(params).get().getOutcome());
+        assertEquals(PolicyResult.FAILURE_TIMEOUT, oper.startOperation(params).get().getResult());
     }
 
     /**
@@ -466,40 +453,45 @@ public class OperatorPartialTest {
         // trigger timeout very quickly
         oper = new MyOper() {
             @Override
-            protected long getTimeOutMillis(Policy policy) {
+            protected long getTimeOutMillis(Integer timeoutSec) {
                 return 10;
             }
 
             @Override
-            protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doPreprocessorAsFuture(
-                            ControlLoopOperationParams params) {
-
-                return outcome -> {
-                    outcome.setOutcome(PolicyResult.SUCCESS.toString());
-
-                    /*
-                     * Create an incomplete future that will timeout after the operation's
-                     * timeout. If it fires before the other timer, then it will return a
-                     * SUCCESS outcome.
-                     */
-                    CompletableFuture<ControlLoopOperation> future = new CompletableFuture<>();
-                    future = future.orTimeout(200, TimeUnit.MILLISECONDS).handleAsync((unused1, unused2) -> outcome,
-                                    params.getExecutor());
-
-                    return future;
+            protected Executor getBlockingExecutor() {
+                return command -> {
+                    Thread thread = new Thread(command);
+                    thread.start();
                 };
             }
+
+            @Override
+            protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
+
+                OperationOutcome outcome = makeSuccess();
+
+                /*
+                 * Create an incomplete future that will timeout after the operation's
+                 * timeout. If it fires before the other timer, then it will return a
+                 * SUCCESS outcome.
+                 */
+                CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
+                future = future.orTimeout(200, TimeUnit.MILLISECONDS).handleAsync((unused1, unused2) -> outcome,
+                                params.getExecutor());
+
+                return future;
+            }
         };
 
         oper.configure(new TreeMap<>());
         oper.start();
 
-        ControlLoopOperation result = oper.startOperation(params).get();
-        assertEquals(PolicyResult.SUCCESS.toString(), result.getOutcome());
+        OperationOutcome result = oper.startOperation(params).get();
+        assertEquals(PolicyResult.SUCCESS, result.getResult());
 
         assertNotNull(opstart);
         assertNotNull(opend);
-        assertEquals(PolicyResult.SUCCESS.toString(), opend.getOutcome());
+        assertEquals(PolicyResult.SUCCESS, opend.getResult());
 
         assertEquals(1, numStart);
         assertEquals(1, oper.getCount());
@@ -510,8 +502,8 @@ public class OperatorPartialTest {
      * Tests retry functions, when the count is set to zero and retries are exhausted.
      */
     @Test
-    public void testSetRetryFlag_testRetryOnFailure_ZeroRetries() {
-        policy.setRetry(0);
+    public void testSetRetryFlag_testRetryOnFailure_ZeroRetries_testStartOperationAttempt() {
+        params = params.toBuilder().retry(0).build();
         oper.setMaxFailures(10);
 
         verifyRun("testSetRetryFlag_testRetryOnFailure_ZeroRetries", 1, 1, PolicyResult.FAILURE);
@@ -522,7 +514,7 @@ public class OperatorPartialTest {
      */
     @Test
     public void testSetRetryFlag_testRetryOnFailure_NullRetries() {
-        policy.setRetry(null);
+        params = params.toBuilder().retry(null).build();
         oper.setMaxFailures(10);
 
         verifyRun("testSetRetryFlag_testRetryOnFailure_NullRetries", 1, 1, PolicyResult.FAILURE);
@@ -534,10 +526,11 @@ public class OperatorPartialTest {
     @Test
     public void testSetRetryFlag_testRetryOnFailure_RetriesExhausted() {
         final int maxRetries = 3;
-        policy.setRetry(maxRetries);
+        params = params.toBuilder().retry(maxRetries).build();
         oper.setMaxFailures(10);
 
-        verifyRun("testVerifyRunningWhenNot", maxRetries + 1, maxRetries + 1, PolicyResult.FAILURE_RETRIES);
+        verifyRun("testSetRetryFlag_testRetryOnFailure_RetriesExhausted", maxRetries + 1, maxRetries + 1,
+                        PolicyResult.FAILURE_RETRIES);
     }
 
     /**
@@ -545,7 +538,7 @@ public class OperatorPartialTest {
      */
     @Test
     public void testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries() {
-        policy.setRetry(10);
+        params = params.toBuilder().retry(10).build();
 
         final int maxFailures = 3;
         oper.setMaxFailures(maxFailures);
@@ -563,8 +556,8 @@ public class OperatorPartialTest {
         // arrange to return null from doOperation()
         oper = new MyOper() {
             @Override
-            protected ControlLoopOperation doOperation(ControlLoopOperationParams params, int attempt,
-                            ControlLoopOperation operation) {
+            protected OperationOutcome doOperation(ControlLoopOperationParams params, int attempt,
+                            OperationOutcome operation) {
 
                 // update counters
                 super.doOperation(params, attempt, operation);
@@ -579,96 +572,55 @@ public class OperatorPartialTest {
     }
 
     @Test
-    public void testGetActorOutcome() {
-        assertNull(oper.getActorOutcome(null));
+    public void testIsSameOperation() {
+        assertFalse(oper.isSameOperation(null));
 
-        ControlLoopOperation outcome = params.makeOutcome();
-        outcome.setOutcome(TARGET);
+        OperationOutcome outcome = params.makeOutcome();
 
-        // wrong actor - should be null
+        // wrong actor - should be false
         outcome.setActor(null);
-        assertNull(oper.getActorOutcome(outcome));
+        assertFalse(oper.isSameOperation(outcome));
         outcome.setActor(TARGET);
-        assertNull(oper.getActorOutcome(outcome));
+        assertFalse(oper.isSameOperation(outcome));
         outcome.setActor(ACTOR);
 
         // wrong operation - should be null
         outcome.setOperation(null);
-        assertNull(oper.getActorOutcome(outcome));
+        assertFalse(oper.isSameOperation(outcome));
         outcome.setOperation(TARGET);
-        assertNull(oper.getActorOutcome(outcome));
+        assertFalse(oper.isSameOperation(outcome));
         outcome.setOperation(OPERATOR);
 
-        assertEquals(TARGET, oper.getActorOutcome(outcome));
-    }
-
-    @Test
-    public void testOnSuccess() throws Exception {
-        AtomicInteger count = new AtomicInteger();
-
-        final Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> nextStep = oper -> {
-            count.incrementAndGet();
-            return CompletableFuture.completedFuture(oper);
-        };
-
-        // pass it a null outcome
-        ControlLoopOperation outcome = oper.onSuccess(params, nextStep).apply(null).get();
-        assertNotNull(outcome);
-        assertEquals(PolicyResult.FAILURE.toString(), outcome.getOutcome());
-        assertEquals(0, count.get());
-
-        // pass it an unpopulated (i.e., failed) outcome
-        outcome = new ControlLoopOperation();
-        assertSame(outcome, oper.onSuccess(params, nextStep).apply(outcome).get());
-        assertEquals(0, count.get());
-
-        // pass it a successful outcome
-        outcome = params.makeOutcome();
-        outcome.setOutcome(PolicyResult.SUCCESS.toString());
-        assertSame(outcome, oper.onSuccess(params, nextStep).apply(outcome).get());
-        assertEquals(PolicyResult.SUCCESS.toString(), outcome.getOutcome());
-        assertEquals(1, count.get());
+        assertTrue(oper.isSameOperation(outcome));
     }
 
     /**
-     * Tests onSuccess() and handleFailure() when the outcome is a success.
+     * Tests handleFailure() when the outcome is a success.
      */
     @Test
-    public void testOnSuccessTrue_testHandleFailureTrue() {
-        // arrange to return a success from the preprocessor
-        oper.setPreProcessor(oper -> {
-            oper.setOutcome(PolicyResult.SUCCESS.toString());
-            return CompletableFuture.completedFuture(oper);
-        });
-
-        verifyRun("testOnSuccessTrue_testHandleFailureTrue", 1, 1, PolicyResult.SUCCESS);
+    public void testHandlePreprocessorFailureTrue() {
+        oper.setPreProcessor(CompletableFuture.completedFuture(makeSuccess()));
+        verifyRun("testHandlePreprocessorFailureTrue", 1, 1, PolicyResult.SUCCESS);
     }
 
     /**
-     * Tests onSuccess() and handleFailure() when the outcome is <i>not</i> a success.
+     * Tests handleFailure() when the outcome is <i>not</i> a success.
      */
     @Test
-    public void testOnSuccessFalse_testHandleFailureFalse() throws Exception {
-        // arrange to return a failure from the preprocessor
-        oper.setPreProcessor(oper -> {
-            oper.setOutcome(PolicyResult.FAILURE.toString());
-            return CompletableFuture.completedFuture(oper);
-        });
-
-        verifyRun("testOnSuccessFalse_testHandleFailureFalse", 1, 0, PolicyResult.FAILURE_GUARD);
+    public void testHandlePreprocessorFailureFalse() throws Exception {
+        oper.setPreProcessor(CompletableFuture.completedFuture(makeFailure()));
+        verifyRun("testHandlePreprocessorFailureFalse", 1, 0, PolicyResult.FAILURE_GUARD);
     }
 
     /**
-     * Tests onSuccess() and handleFailure() when the outcome is {@code null}.
+     * Tests handleFailure() when the outcome is {@code null}.
      */
     @Test
-    public void testOnSuccessFalse_testHandleFailureNull() throws Exception {
+    public void testHandlePreprocessorFailureNull() throws Exception {
         // arrange to return null from the preprocessor
-        oper.setPreProcessor(oper -> {
-            return CompletableFuture.completedFuture(null);
-        });
+        oper.setPreProcessor(CompletableFuture.completedFuture(null));
 
-        verifyRun("testOnSuccessFalse_testHandleFailureNull", 1, 0, PolicyResult.FAILURE_GUARD);
+        verifyRun("testHandlePreprocessorFailureNull", 1, 0, PolicyResult.FAILURE_GUARD);
     }
 
     @Test
@@ -688,11 +640,374 @@ public class OperatorPartialTest {
     }
 
     /**
-     * Tests verifyRunning() when the pipeline is not running.
+     * Tests both flavors of anyOf(), because one invokes the other.
+     */
+    @Test
+    public void testAnyOf() throws Exception {
+        // first task completes, others do not
+        List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+        final OperationOutcome outcome = params.makeOutcome();
+
+        tasks.add(CompletableFuture.completedFuture(outcome));
+        tasks.add(new CompletableFuture<>());
+        tasks.add(new CompletableFuture<>());
+
+        CompletableFuture<OperationOutcome> result = oper.anyOf(params, tasks);
+        assertTrue(executor.runAll());
+
+        assertTrue(result.isDone());
+        assertSame(outcome, result.get());
+
+        // second task completes, others do not
+        tasks = new LinkedList<>();
+
+        tasks.add(new CompletableFuture<>());
+        tasks.add(CompletableFuture.completedFuture(outcome));
+        tasks.add(new CompletableFuture<>());
+
+        result = oper.anyOf(params, tasks);
+        assertTrue(executor.runAll());
+
+        assertTrue(result.isDone());
+        assertSame(outcome, result.get());
+
+        // third task completes, others do not
+        tasks = new LinkedList<>();
+
+        tasks.add(new CompletableFuture<>());
+        tasks.add(new CompletableFuture<>());
+        tasks.add(CompletableFuture.completedFuture(outcome));
+
+        result = oper.anyOf(params, tasks);
+        assertTrue(executor.runAll());
+
+        assertTrue(result.isDone());
+        assertSame(outcome, result.get());
+    }
+
+    /**
+     * Tests both flavors of allOf(), because one invokes the other.
+     */
+    @Test
+    public void testAllOf() throws Exception {
+        List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+        final OperationOutcome outcome = params.makeOutcome();
+
+        CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
+        CompletableFuture<OperationOutcome> future2 = new CompletableFuture<>();
+        CompletableFuture<OperationOutcome> future3 = new CompletableFuture<>();
+
+        tasks.add(future1);
+        tasks.add(future2);
+        tasks.add(future3);
+
+        CompletableFuture<OperationOutcome> result = oper.allOf(params, tasks);
+
+        assertTrue(executor.runAll());
+        assertFalse(result.isDone());
+        future1.complete(outcome);
+
+        // complete 3 before 2
+        assertTrue(executor.runAll());
+        assertFalse(result.isDone());
+        future3.complete(outcome);
+
+        assertTrue(executor.runAll());
+        assertFalse(result.isDone());
+        future2.complete(outcome);
+
+        // all of them are now done
+        assertTrue(executor.runAll());
+        assertTrue(result.isDone());
+        assertSame(outcome, result.get());
+    }
+
+    @Test
+    public void testCombineOutcomes() throws Exception {
+        // only one outcome
+        verifyOutcomes(0, PolicyResult.SUCCESS);
+        verifyOutcomes(0, PolicyResult.FAILURE_EXCEPTION);
+
+        // maximum is in different positions
+        verifyOutcomes(0, PolicyResult.FAILURE, PolicyResult.SUCCESS, PolicyResult.FAILURE_GUARD);
+        verifyOutcomes(1, PolicyResult.SUCCESS, PolicyResult.FAILURE, PolicyResult.FAILURE_GUARD);
+        verifyOutcomes(2, PolicyResult.SUCCESS, PolicyResult.FAILURE_GUARD, PolicyResult.FAILURE);
+
+        // null outcome
+        final List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+        tasks.add(CompletableFuture.completedFuture(null));
+        CompletableFuture<OperationOutcome> result = oper.allOf(params, tasks);
+
+        assertTrue(executor.runAll());
+        assertTrue(result.isDone());
+        assertNull(result.get());
+
+        // one throws an exception during execution
+        IllegalStateException except = new IllegalStateException(EXPECTED_EXCEPTION);
+
+        tasks.clear();
+        tasks.add(CompletableFuture.completedFuture(params.makeOutcome()));
+        tasks.add(CompletableFuture.failedFuture(except));
+        tasks.add(CompletableFuture.completedFuture(params.makeOutcome()));
+        result = oper.allOf(params, tasks);
+
+        assertTrue(executor.runAll());
+        assertTrue(result.isCompletedExceptionally());
+        result.whenComplete((unused, thrown) -> assertSame(except, thrown));
+    }
+
+    private void verifyOutcomes(int expected, PolicyResult... results) throws Exception {
+        List<CompletableFuture<OperationOutcome>> tasks = new LinkedList<>();
+
+
+        OperationOutcome expectedOutcome = null;
+
+        for (int count = 0; count < results.length; ++count) {
+            OperationOutcome outcome = params.makeOutcome();
+            outcome.setResult(results[count]);
+            tasks.add(CompletableFuture.completedFuture(outcome));
+
+            if (count == expected) {
+                expectedOutcome = outcome;
+            }
+        }
+
+        CompletableFuture<OperationOutcome> result = oper.allOf(params, tasks);
+
+        assertTrue(executor.runAll());
+        assertTrue(result.isDone());
+        assertSame(expectedOutcome, result.get());
+    }
+
+    private Function<OperationOutcome, CompletableFuture<OperationOutcome>> makeTask(
+                    final OperationOutcome taskOutcome) {
+
+        return outcome -> CompletableFuture.completedFuture(taskOutcome);
+    }
+
+    @Test
+    public void testDetmPriority() {
+        assertEquals(1, oper.detmPriority(null));
+
+        OperationOutcome outcome = params.makeOutcome();
+
+        Map<PolicyResult, Integer> map = Map.of(PolicyResult.SUCCESS, 0, PolicyResult.FAILURE_GUARD, 2,
+                        PolicyResult.FAILURE_RETRIES, 3, PolicyResult.FAILURE, 4, PolicyResult.FAILURE_TIMEOUT, 5,
+                        PolicyResult.FAILURE_EXCEPTION, 6);
+
+        for (Entry<PolicyResult, Integer> ent : map.entrySet()) {
+            outcome.setResult(ent.getKey());
+            assertEquals(ent.getKey().toString(), ent.getValue().intValue(), oper.detmPriority(outcome));
+        }
+    }
+
+    /**
+     * Tests doTask(Future) when the controller is not running.
+     */
+    @Test
+    public void testDoTaskFutureNotRunning() throws Exception {
+        CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+        controller.complete(params.makeOutcome());
+
+        CompletableFuture<OperationOutcome> future =
+                        oper.doTask(params, controller, false, params.makeOutcome(), taskFuture);
+        assertFalse(future.isDone());
+        assertTrue(executor.runAll());
+
+        // should not have run the task
+        assertFalse(future.isDone());
+
+        // should have canceled the task future
+        assertTrue(taskFuture.isCancelled());
+    }
+
+    /**
+     * Tests doTask(Future) when the previous outcome was successful.
+     */
+    @Test
+    public void testDoTaskFutureSuccess() throws Exception {
+        CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+        final OperationOutcome taskOutcome = params.makeOutcome();
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+        CompletableFuture<OperationOutcome> future =
+                        oper.doTask(params, controller, true, params.makeOutcome(), taskFuture);
+
+        taskFuture.complete(taskOutcome);
+        assertTrue(executor.runAll());
+
+        assertTrue(future.isDone());
+        assertSame(taskOutcome, future.get());
+
+        // controller should not be done yet
+        assertFalse(controller.isDone());
+    }
+
+    /**
+     * Tests doTask(Future) when the previous outcome was failed.
+     */
+    @Test
+    public void testDoTaskFutureFailure() throws Exception {
+        CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+        final OperationOutcome failedOutcome = params.makeOutcome();
+        failedOutcome.setResult(PolicyResult.FAILURE);
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+        CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, true, failedOutcome, taskFuture);
+        assertFalse(future.isDone());
+        assertTrue(executor.runAll());
+
+        // should not have run the task
+        assertFalse(future.isDone());
+
+        // should have canceled the task future
+        assertTrue(taskFuture.isCancelled());
+
+        // controller SHOULD be done now
+        assertTrue(controller.isDone());
+        assertSame(failedOutcome, controller.get());
+    }
+
+    /**
+     * Tests doTask(Future) when the previous outcome was failed, but not checking
+     * success.
+     */
+    @Test
+    public void testDoTaskFutureUncheckedFailure() throws Exception {
+        CompletableFuture<OperationOutcome> taskFuture = new CompletableFuture<>();
+        final OperationOutcome failedOutcome = params.makeOutcome();
+        failedOutcome.setResult(PolicyResult.FAILURE);
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+        CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, false, failedOutcome, taskFuture);
+        assertFalse(future.isDone());
+
+        // complete the task
+        OperationOutcome taskOutcome = params.makeOutcome();
+        taskFuture.complete(taskOutcome);
+
+        assertTrue(executor.runAll());
+
+        // should have run the task
+        assertTrue(future.isDone());
+
+        assertTrue(future.isDone());
+        assertSame(taskOutcome, future.get());
+
+        // controller should not be done yet
+        assertFalse(controller.isDone());
+    }
+
+    /**
+     * Tests doTask(Function) when the controller is not running.
+     */
+    @Test
+    public void testDoTaskFunctionNotRunning() throws Exception {
+        AtomicBoolean invoked = new AtomicBoolean();
+
+        Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = outcome -> {
+            invoked.set(true);
+            return CompletableFuture.completedFuture(params.makeOutcome());
+        };
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+        controller.complete(params.makeOutcome());
+
+        CompletableFuture<OperationOutcome> future =
+                        oper.doTask(params, controller, false, task).apply(params.makeOutcome());
+        assertFalse(future.isDone());
+        assertTrue(executor.runAll());
+
+        // should not have run the task
+        assertFalse(future.isDone());
+
+        // should not have even invoked the task
+        assertFalse(invoked.get());
+    }
+
+    /**
+     * Tests doTask(Function) when the previous outcome was successful.
+     */
+    @Test
+    public void testDoTaskFunctionSuccess() throws Exception {
+        final OperationOutcome taskOutcome = params.makeOutcome();
+
+        final OperationOutcome failedOutcome = params.makeOutcome();
+
+        Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = makeTask(taskOutcome);
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+        CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, true, task).apply(failedOutcome);
+
+        assertTrue(future.isDone());
+        assertSame(taskOutcome, future.get());
+
+        // controller should not be done yet
+        assertFalse(controller.isDone());
+    }
+
+    /**
+     * Tests doTask(Function) when the previous outcome was failed.
+     */
+    @Test
+    public void testDoTaskFunctionFailure() throws Exception {
+        final OperationOutcome failedOutcome = params.makeOutcome();
+        failedOutcome.setResult(PolicyResult.FAILURE);
+
+        AtomicBoolean invoked = new AtomicBoolean();
+
+        Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = outcome -> {
+            invoked.set(true);
+            return CompletableFuture.completedFuture(params.makeOutcome());
+        };
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+        CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, true, task).apply(failedOutcome);
+        assertFalse(future.isDone());
+        assertTrue(executor.runAll());
+
+        // should not have run the task
+        assertFalse(future.isDone());
+
+        // should not have even invoked the task
+        assertFalse(invoked.get());
+
+        // controller should have the failed task
+        assertTrue(controller.isDone());
+        assertSame(failedOutcome, controller.get());
+    }
+
+    /**
+     * Tests doTask(Function) when the previous outcome was failed, but not checking
+     * success.
      */
     @Test
-    public void testVerifyRunningWhenNot() {
-        verifyRun("testVerifyRunningWhenNot", 0, 0, PolicyResult.SUCCESS, future -> future.cancel(false));
+    public void testDoTaskFunctionUncheckedFailure() throws Exception {
+        final OperationOutcome taskOutcome = params.makeOutcome();
+
+        final OperationOutcome failedOutcome = params.makeOutcome();
+        failedOutcome.setResult(PolicyResult.FAILURE);
+
+        Function<OperationOutcome, CompletableFuture<OperationOutcome>> task = makeTask(taskOutcome);
+
+        PipelineControllerFuture<OperationOutcome> controller = new PipelineControllerFuture<>();
+
+        CompletableFuture<OperationOutcome> future = oper.doTask(params, controller, false, task).apply(failedOutcome);
+
+        assertTrue(future.isDone());
+        assertSame(taskOutcome, future.get());
+
+        // controller should not be done yet
+        assertFalse(controller.isDone());
     }
 
     /**
@@ -700,7 +1015,7 @@ public class OperatorPartialTest {
      */
     @Test
     public void testCallbackStartedNotRunning() {
-        AtomicReference<Future<ControlLoopOperation>> future = new AtomicReference<>();
+        AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
 
         /*
          * arrange to stop the controller when the start-callback is invoked, but capture
@@ -723,7 +1038,7 @@ public class OperatorPartialTest {
      */
     @Test
     public void testCallbackCompletedNotRunning() {
-        AtomicReference<Future<ControlLoopOperation>> future = new AtomicReference<>();
+        AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
 
         // arrange to stop the controller when the start-callback is invoked
         params = params.toBuilder().startCallback(oper -> {
@@ -739,36 +1054,36 @@ public class OperatorPartialTest {
     }
 
     @Test
-    public void testSetOutcomeControlLoopOperationThrowable() {
+    public void testSetOutcomeControlLoopOperationOutcomeThrowable() {
         final CompletionException timex = new CompletionException(new TimeoutException(EXPECTED_EXCEPTION));
 
-        ControlLoopOperation outcome;
+        OperationOutcome outcome;
 
-        outcome = new ControlLoopOperation();
+        outcome = new OperationOutcome();
         oper.setOutcome(params, outcome, timex);
         assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
-        assertEquals(PolicyResult.FAILURE_TIMEOUT.toString(), outcome.getOutcome());
+        assertEquals(PolicyResult.FAILURE_TIMEOUT, outcome.getResult());
 
-        outcome = new ControlLoopOperation();
-        oper.setOutcome(params, outcome, new IllegalStateException());
+        outcome = new OperationOutcome();
+        oper.setOutcome(params, outcome, new IllegalStateException(EXPECTED_EXCEPTION));
         assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
-        assertEquals(PolicyResult.FAILURE_EXCEPTION.toString(), outcome.getOutcome());
+        assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
     }
 
     @Test
-    public void testSetOutcomeControlLoopOperationPolicyResult() {
-        ControlLoopOperation outcome;
+    public void testSetOutcomeControlLoopOperationOutcomePolicyResult() {
+        OperationOutcome outcome;
 
-        outcome = new ControlLoopOperation();
+        outcome = new OperationOutcome();
         oper.setOutcome(params, outcome, PolicyResult.SUCCESS);
         assertEquals(ControlLoopOperation.SUCCESS_MSG, outcome.getMessage());
-        assertEquals(PolicyResult.SUCCESS.toString(), outcome.getOutcome());
+        assertEquals(PolicyResult.SUCCESS, outcome.getResult());
 
         for (PolicyResult result : FAILURE_RESULTS) {
-            outcome = new ControlLoopOperation();
+            outcome = new OperationOutcome();
             oper.setOutcome(params, outcome, result);
             assertEquals(result.toString(), ControlLoopOperation.FAILED_MSG, outcome.getMessage());
-            assertEquals(result.toString(), result.toString(), outcome.getOutcome());
+            assertEquals(result.toString(), result, outcome.getResult());
         }
     }
 
@@ -776,7 +1091,7 @@ public class OperatorPartialTest {
     public void testIsTimeout() {
         final TimeoutException timex = new TimeoutException(EXPECTED_EXCEPTION);
 
-        assertFalse(oper.isTimeout(new IllegalStateException()));
+        assertFalse(oper.isTimeout(new IllegalStateException(EXPECTED_EXCEPTION)));
         assertFalse(oper.isTimeout(new IllegalStateException(timex)));
         assertFalse(oper.isTimeout(new CompletionException(new IllegalStateException(timex))));
         assertFalse(oper.isTimeout(new CompletionException(null)));
@@ -788,19 +1103,19 @@ public class OperatorPartialTest {
 
     @Test
     public void testGetTimeOutMillis() {
-        assertEquals(TIMEOUT * 1000, oper.getTimeOutMillis(policy));
+        assertEquals(TIMEOUT * 1000, oper.getTimeOutMillis(params.getTimeoutSec()));
 
-        policy.setTimeout(null);
-        assertEquals(0, oper.getTimeOutMillis(policy));
+        params = params.toBuilder().timeoutSec(null).build();
+        assertEquals(0, oper.getTimeOutMillis(params.getTimeoutSec()));
     }
 
-    private void starter(ControlLoopOperation oper) {
+    private void starter(OperationOutcome oper) {
         ++numStart;
         tstart = oper.getStart();
         opstart = oper;
     }
 
-    private void completer(ControlLoopOperation oper) {
+    private void completer(OperationOutcome oper) {
         ++numEnd;
         opend = oper;
     }
@@ -816,21 +1131,18 @@ public class OperatorPartialTest {
         };
     }
 
-    /**
-     * Verifies a run.
-     *
-     * @param testName test name
-     * @param expectedCallbacks number of callbacks expected
-     * @param expectedOperations number of operation invocations expected
-     * @param expectedResult expected outcome
-     */
-    private void verifyRun(String testName, int expectedCallbacks, int expectedOperations,
-                    PolicyResult expectedResult) {
+    private OperationOutcome makeSuccess() {
+        OperationOutcome outcome = params.makeOutcome();
+        outcome.setResult(PolicyResult.SUCCESS);
 
-        String expectedSubRequestId =
-                        (expectedResult == PolicyResult.FAILURE_EXCEPTION ? null : String.valueOf(expectedOperations));
+        return outcome;
+    }
 
-        verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, noop());
+    private OperationOutcome makeFailure() {
+        OperationOutcome outcome = params.makeOutcome();
+        outcome.setResult(PolicyResult.FAILURE);
+
+        return outcome;
     }
 
     /**
@@ -840,17 +1152,14 @@ public class OperatorPartialTest {
      * @param expectedCallbacks number of callbacks expected
      * @param expectedOperations number of operation invocations expected
      * @param expectedResult expected outcome
-     * @param manipulator function to modify the future returned by
-     *        {@link OperatorPartial#startOperation(ControlLoopOperationParams)} before
-     *        the tasks in the executor are run
      */
-    private void verifyRun(String testName, int expectedCallbacks, int expectedOperations, PolicyResult expectedResult,
-                    Consumer<CompletableFuture<ControlLoopOperation>> manipulator) {
+    private void verifyRun(String testName, int expectedCallbacks, int expectedOperations,
+                    PolicyResult expectedResult) {
 
         String expectedSubRequestId =
                         (expectedResult == PolicyResult.FAILURE_EXCEPTION ? null : String.valueOf(expectedOperations));
 
-        verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, manipulator);
+        verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, expectedSubRequestId, noop());
     }
 
     /**
@@ -866,9 +1175,9 @@ public class OperatorPartialTest {
      *        the tasks in the executor are run
      */
     private void verifyRun(String testName, int expectedCallbacks, int expectedOperations, PolicyResult expectedResult,
-                    String expectedSubRequestId, Consumer<CompletableFuture<ControlLoopOperation>> manipulator) {
+                    String expectedSubRequestId, Consumer<CompletableFuture<OperationOutcome>> manipulator) {
 
-        CompletableFuture<ControlLoopOperation> future = oper.startOperation(params);
+        CompletableFuture<OperationOutcome> future = oper.startOperation(params);
 
         manipulator.accept(future);
 
@@ -880,7 +1189,7 @@ public class OperatorPartialTest {
         if (expectedCallbacks > 0) {
             assertNotNull(testName, opstart);
             assertNotNull(testName, opend);
-            assertEquals(testName, expectedResult.toString(), opend.getOutcome());
+            assertEquals(testName, expectedResult, opend.getResult());
 
             assertSame(testName, tstart, opstart.getStart());
             assertSame(testName, tstart, opend.getStart());
@@ -901,7 +1210,7 @@ public class OperatorPartialTest {
         assertEquals(testName, expectedOperations, oper.getCount());
     }
 
-    private static class MyOper extends OperatorPartial {
+    private class MyOper extends OperatorPartial {
         @Getter
         private int count = 0;
 
@@ -912,15 +1221,15 @@ public class OperatorPartialTest {
         private int maxFailures = 0;
 
         @Setter
-        private Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> preProcessor;
+        private CompletableFuture<OperationOutcome> preProcessor;
 
         public MyOper() {
             super(ACTOR, OPERATOR);
         }
 
         @Override
-        protected ControlLoopOperation doOperation(ControlLoopOperationParams params, int attempt,
-                        ControlLoopOperation operation) {
+        protected OperationOutcome doOperation(ControlLoopOperationParams params, int attempt,
+                        OperationOutcome operation) {
             ++count;
             if (genException) {
                 throw new IllegalStateException(EXPECTED_EXCEPTION);
@@ -929,19 +1238,22 @@ public class OperatorPartialTest {
             operation.setSubRequestId(String.valueOf(attempt));
 
             if (count > maxFailures) {
-                operation.setOutcome(PolicyResult.SUCCESS.toString());
+                operation.setResult(PolicyResult.SUCCESS);
             } else {
-                operation.setOutcome(PolicyResult.FAILURE.toString());
+                operation.setResult(PolicyResult.FAILURE);
             }
 
             return operation;
         }
 
         @Override
-        protected Function<ControlLoopOperation, CompletableFuture<ControlLoopOperation>> doPreprocessorAsFuture(
-                        ControlLoopOperationParams params) {
+        protected CompletableFuture<OperationOutcome> startPreprocessorAsync(ControlLoopOperationParams params) {
+            return (preProcessor != null ? preProcessor : super.startPreprocessorAsync(params));
+        }
 
-            return (preProcessor != null ? preProcessor : super.doPreprocessorAsFuture(params));
+        @Override
+        protected Executor getBlockingExecutor() {
+            return executor;
         }
     }
 
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java
index 0c8e77d38..9dd19d548 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/ControlLoopOperationParamsTest.java
@@ -35,6 +35,8 @@ import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.util.Map;
+import java.util.TreeMap;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
@@ -47,20 +49,23 @@ import org.junit.Test;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 import org.onap.policy.common.parameters.BeanValidationResult;
-import org.onap.policy.controlloop.ControlLoopOperation;
 import org.onap.policy.controlloop.VirtualControlLoopEvent;
 import org.onap.policy.controlloop.actorserviceprovider.ActorService;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
 import org.onap.policy.controlloop.actorserviceprovider.Operator;
 import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams.ControlLoopOperationParamsBuilder;
 import org.onap.policy.controlloop.actorserviceprovider.spi.Actor;
-import org.onap.policy.controlloop.policy.Policy;
+import org.onap.policy.controlloop.policy.Target;
 
 public class ControlLoopOperationParamsTest {
     private static final String EXPECTED_EXCEPTION = "expected exception";
     private static final String ACTOR = "my-actor";
     private static final String OPERATION = "my-operation";
-    private static final String TARGET = "my-target";
+    private static final Target TARGET = new Target();
+    private static final String TARGET_ENTITY = "my-target";
+    private static final Integer RETRY = 3;
+    private static final Integer TIMEOUT = 100;
     private static final UUID REQ_ID = UUID.randomUUID();
 
     @Mock
@@ -70,7 +75,7 @@ public class ControlLoopOperationParamsTest {
     private ActorService actorService;
 
     @Mock
-    private Consumer<ControlLoopOperation> completer;
+    private Consumer<OperationOutcome> completer;
 
     @Mock
     private ControlLoopEventContext context;
@@ -82,19 +87,18 @@ public class ControlLoopOperationParamsTest {
     private Executor executor;
 
     @Mock
-    private CompletableFuture<ControlLoopOperation> operation;
+    private CompletableFuture<OperationOutcome> operation;
 
     @Mock
     private Operator operator;
 
     @Mock
-    private Policy policy;
+    private Consumer<OperationOutcome> starter;
 
-    @Mock
-    private Consumer<ControlLoopOperation> starter;
+    private Map<String, String> payload;
 
     private ControlLoopOperationParams params;
-    private ControlLoopOperation outcome;
+    private OperationOutcome outcome;
 
 
     /**
@@ -112,12 +116,12 @@ public class ControlLoopOperationParamsTest {
 
         when(context.getEvent()).thenReturn(event);
 
-        when(policy.getActor()).thenReturn(ACTOR);
-        when(policy.getRecipe()).thenReturn(OPERATION);
+        payload = new TreeMap<>();
 
         params = ControlLoopOperationParams.builder().actorService(actorService).completeCallback(completer)
-                        .context(context).executor(executor).policy(policy).startCallback(starter).target(TARGET)
-                        .build();
+                        .context(context).executor(executor).actor(ACTOR).operation(OPERATION).payload(payload)
+                        .retry(RETRY).target(TARGET).targetEntity(TARGET_ENTITY).timeoutSec(TIMEOUT)
+                        .startCallback(starter).build();
 
         outcome = params.makeOutcome();
     }
@@ -130,30 +134,6 @@ public class ControlLoopOperationParamsTest {
     }
 
     @Test
-    public void testGetActor() {
-        assertEquals(ACTOR, params.getActor());
-
-        // try with null policy
-        assertEquals(ControlLoopOperationParams.UNKNOWN, params.toBuilder().policy(null).build().getActor());
-
-        // try with null name in the policy
-        when(policy.getActor()).thenReturn(null);
-        assertEquals(ControlLoopOperationParams.UNKNOWN, params.getActor());
-    }
-
-    @Test
-    public void testGetOperation() {
-        assertEquals(OPERATION, params.getOperation());
-
-        // try with null policy
-        assertEquals(ControlLoopOperationParams.UNKNOWN, params.toBuilder().policy(null).build().getOperation());
-
-        // try with null name in the policy
-        when(policy.getRecipe()).thenReturn(null);
-        assertEquals(ControlLoopOperationParams.UNKNOWN, params.getOperation());
-    }
-
-    @Test
     public void testGetRequestId() {
         assertSame(REQ_ID, params.getRequestId());
 
@@ -170,20 +150,14 @@ public class ControlLoopOperationParamsTest {
         assertEquals(ACTOR, outcome.getActor());
         assertEquals(OPERATION, outcome.getOperation());
         checkRemainingFields("with actor");
-
-        // try again with a null policy
-        outcome = params.toBuilder().policy(null).build().makeOutcome();
-        assertEquals(ControlLoopOperationParams.UNKNOWN, outcome.getActor());
-        assertEquals(ControlLoopOperationParams.UNKNOWN, outcome.getOperation());
-        checkRemainingFields("unknown actor");
     }
 
     protected void checkRemainingFields(String testName) {
-        assertEquals(testName, TARGET, outcome.getTarget());
-        assertNotNull(testName, outcome.getStart());
+        assertEquals(testName, TARGET_ENTITY, outcome.getTarget());
+        assertNull(testName, outcome.getStart());
         assertNull(testName, outcome.getEnd());
         assertNull(testName, outcome.getSubRequestId());
-        assertNull(testName, outcome.getOutcome());
+        assertNotNull(testName, outcome.getResult());
         assertNull(testName, outcome.getMessage());
     }
 
@@ -239,21 +213,23 @@ public class ControlLoopOperationParamsTest {
 
     @Test
     public void testValidateFields() {
+        testValidate("actor", "null", bldr -> bldr.actor(null));
         testValidate("actorService", "null", bldr -> bldr.actorService(null));
         testValidate("context", "null", bldr -> bldr.context(null));
         testValidate("executor", "null", bldr -> bldr.executor(null));
-        testValidate("policy", "null", bldr -> bldr.policy(null));
-        testValidate("target", "null", bldr -> bldr.target(null));
+        testValidate("operation", "null", bldr -> bldr.operation(null));
+        testValidate("target", "null", bldr -> bldr.targetEntity(null));
 
         // check edge cases
         assertTrue(params.toBuilder().build().validate().isValid());
 
         // these can be null
-        assertTrue(params.toBuilder().startCallback(null).completeCallback(null).build().validate().isValid());
+        assertTrue(params.toBuilder().payload(null).retry(null).target(null).timeoutSec(null).startCallback(null)
+                        .completeCallback(null).build().validate().isValid());
 
         // test with minimal fields
-        assertTrue(ControlLoopOperationParams.builder().actorService(actorService).context(context).policy(policy)
-                        .target(TARGET).build().validate().isValid());
+        assertTrue(ControlLoopOperationParams.builder().actorService(actorService).context(context).actor(ACTOR)
+                        .operation(OPERATION).targetEntity(TARGET_ENTITY).build().validate().isValid());
     }
 
     private void testValidate(String fieldName, String expected,
@@ -275,7 +251,12 @@ public class ControlLoopOperationParamsTest {
     }
 
     @Test
-    public void testActorService() {
+    public void testGetActor() {
+        assertSame(ACTOR, params.getActor());
+    }
+
+    @Test
+    public void testGetActorService() {
         assertSame(actorService, params.getActorService());
     }
 
@@ -293,8 +274,43 @@ public class ControlLoopOperationParamsTest {
     }
 
     @Test
-    public void testGetPolicy() {
-        assertSame(policy, params.getPolicy());
+    public void testGetOperation() {
+        assertSame(OPERATION, params.getOperation());
+    }
+
+    @Test
+    public void testGetPayload() {
+        assertSame(payload, params.getPayload());
+
+        // should be null when unspecified
+        assertNull(ControlLoopOperationParams.builder().build().getPayload());
+    }
+
+    @Test
+    public void testGetRetry() {
+        assertSame(RETRY, params.getRetry());
+
+        // should be null when unspecified
+        assertNull(ControlLoopOperationParams.builder().build().getRetry());
+    }
+
+    @Test
+    public void testTarget() {
+        assertSame(TARGET, params.getTarget());
+
+        // should be null when unspecified
+        assertNull(ControlLoopOperationParams.builder().build().getTarget());
+    }
+
+    @Test
+    public void testGetTimeoutSec() {
+        assertSame(TIMEOUT, params.getTimeoutSec());
+
+        // should be 300 when unspecified
+        assertEquals(Integer.valueOf(300), ControlLoopOperationParams.builder().build().getTimeoutSec());
+
+        // null should be ok too
+        assertNull(ControlLoopOperationParams.builder().timeoutSec(null).build().getTimeoutSec());
     }
 
     @Test
@@ -308,7 +324,7 @@ public class ControlLoopOperationParamsTest {
     }
 
     @Test
-    public void testGetTarget() {
-        assertEquals(TARGET, params.getTarget());
+    public void testGetTargetEntity() {
+        assertEquals(TARGET_ENTITY, params.getTargetEntity());
     }
 }
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java
index 1763388f2..6c1f538ec 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpActorParamsTest.java
@@ -90,6 +90,8 @@ public class HttpActorParamsTest {
 
     @Test
     public void testValidate() {
+        assertTrue(params.validate(CONTAINER).isValid());
+
         testValidateField("clientName", "null", params2 -> params2.setClientName(null));
         testValidateField("path", "null", params2 -> params2.setPath(null));
         testValidateField("timeoutSec", "minimum", params2 -> params2.setTimeoutSec(-1));
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java
index 829c480d1..6cf7328ca 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/parameters/HttpParamsTest.java
@@ -47,6 +47,8 @@ public class HttpParamsTest {
 
     @Test
     public void testValidate() {
+        assertTrue(params.validate(CONTAINER).isValid());
+
         testValidateField("clientName", "null", bldr -> bldr.clientName(null));
         testValidateField("path", "null", bldr -> bldr.path(null));
         testValidateField("timeoutSec", "minimum", bldr -> bldr.timeoutSec(-1));
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java
index b421c1ce2..a6b11ef65 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/pipeline/PipelineControllerFutureTest.java
@@ -23,21 +23,27 @@ package org.onap.policy.controlloop.actorserviceprovider.pipeline;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.Function;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
@@ -58,9 +64,10 @@ public class PipelineControllerFutureTest {
     private Future<String> future2;
 
     @Mock
-    private CompletableFuture<String> compFuture;
+    private Executor executor;
 
 
+    private CompletableFuture<String> compFuture;
     private PipelineControllerFuture<String> controller;
 
 
@@ -72,6 +79,8 @@ public class PipelineControllerFutureTest {
     public void setUp() {
         MockitoAnnotations.initMocks(this);
 
+        compFuture = spy(new CompletableFuture<>());
+
         controller = new PipelineControllerFuture<>();
 
         controller.add(runnable1);
@@ -89,10 +98,7 @@ public class PipelineControllerFutureTest {
         assertTrue(controller.isCancelled());
         assertFalse(controller.isRunning());
 
-        verify(runnable1).run();
-        verify(runnable2).run();
-        verify(future1).cancel(anyBoolean());
-        verify(future2).cancel(anyBoolean());
+        verifyStopped();
 
         // re-invoke; nothing should change
         assertTrue(controller.cancel(true));
@@ -100,10 +106,155 @@ public class PipelineControllerFutureTest {
         assertTrue(controller.isCancelled());
         assertFalse(controller.isRunning());
 
-        verify(runnable1).run();
-        verify(runnable2).run();
-        verify(future1).cancel(anyBoolean());
-        verify(future2).cancel(anyBoolean());
+        verifyStopped();
+    }
+
+    @Test
+    public void testCompleteT() throws Exception {
+        assertTrue(controller.complete(TEXT));
+        assertEquals(TEXT, controller.get());
+
+        verifyStopped();
+
+        // repeat - disallowed
+        assertFalse(controller.complete(TEXT));
+    }
+
+    @Test
+    public void testCompleteExceptionallyThrowable() {
+        assertTrue(controller.completeExceptionally(EXPECTED_EXCEPTION));
+        assertThatThrownBy(() -> controller.get()).hasCause(EXPECTED_EXCEPTION);
+
+        verifyStopped();
+
+        // repeat - disallowed
+        assertFalse(controller.completeExceptionally(EXPECTED_EXCEPTION));
+    }
+
+    @Test
+    public void testCompleteAsyncSupplierOfQextendsTExecutor() throws Exception {
+        CompletableFuture<String> future = controller.completeAsync(() -> TEXT, executor);
+
+        // haven't stopped anything yet
+        assertFalse(future.isDone());
+        verify(runnable1, never()).run();
+
+        // get the operation and run it
+        ArgumentCaptor<Runnable> captor = ArgumentCaptor.forClass(Runnable.class);
+        verify(executor).execute(captor.capture());
+        captor.getValue().run();
+
+        // should be done now
+        assertTrue(future.isDone());
+
+        assertEquals(TEXT, future.get());
+
+        verifyStopped();
+    }
+
+    /**
+     * Tests completeAsync(executor) when canceled before execution.
+     */
+    @Test
+    public void testCompleteAsyncSupplierOfQextendsTExecutorCanceled() throws Exception {
+        CompletableFuture<String> future = controller.completeAsync(() -> TEXT, executor);
+
+        assertTrue(future.cancel(false));
+
+        verifyStopped();
+
+        assertTrue(future.isDone());
+
+        assertThatThrownBy(() -> controller.get()).isInstanceOf(CancellationException.class);
+    }
+
+    @Test
+    public void testCompleteAsyncSupplierOfQextendsT() throws Exception {
+        CompletableFuture<String> future = controller.completeAsync(() -> TEXT);
+        assertEquals(TEXT, future.get());
+
+        verifyStopped();
+    }
+
+    /**
+     * Tests completeAsync() when canceled.
+     */
+    @Test
+    public void testCompleteAsyncSupplierOfQextendsTCanceled() throws Exception {
+        CountDownLatch canceled = new CountDownLatch(1);
+
+        // run async, but await until canceled
+        CompletableFuture<String> future = controller.completeAsync(() -> {
+            try {
+                canceled.await();
+            } catch (InterruptedException e) {
+                // do nothing
+            }
+
+            return TEXT;
+        });
+
+        assertTrue(future.cancel(false));
+
+        // let the future run now
+        canceled.countDown();
+
+        verifyStopped();
+
+        assertTrue(future.isDone());
+
+        assertThatThrownBy(() -> controller.get()).isInstanceOf(CancellationException.class);
+    }
+
+    @Test
+    public void testCompleteOnTimeoutTLongTimeUnit() throws Exception {
+        CountDownLatch stopped = new CountDownLatch(1);
+        controller.add(() -> stopped.countDown());
+
+        CompletableFuture<String> future = controller.completeOnTimeout(TEXT, 1, TimeUnit.MILLISECONDS);
+
+        assertEquals(TEXT, future.get());
+
+        /*
+         * Must use latch instead of verifyStopped(), because the runnables may be
+         * executed asynchronously.
+         */
+        assertTrue(stopped.await(5, TimeUnit.SECONDS));
+    }
+
+    /**
+     * Tests completeOnTimeout() when completed before the timeout.
+     */
+    @Test
+    public void testCompleteOnTimeoutTLongTimeUnitNoTimeout() throws Exception {
+        CompletableFuture<String> future = controller.completeOnTimeout("timed out", 5, TimeUnit.SECONDS);
+        controller.complete(TEXT);
+
+        assertEquals(TEXT, future.get());
+
+        verifyStopped();
+    }
+
+    /**
+     * Tests completeOnTimeout() when canceled before the timeout.
+     */
+    @Test
+    public void testCompleteOnTimeoutTLongTimeUnitCanceled() {
+        CompletableFuture<String> future = controller.completeOnTimeout(TEXT, 5, TimeUnit.SECONDS);
+        assertTrue(future.cancel(true));
+
+        assertThatThrownBy(() -> controller.get()).isInstanceOf(CancellationException.class);
+
+        verifyStopped();
+    }
+
+    @Test
+    public void testNewIncompleteFuture() {
+        PipelineControllerFuture<String> future = controller.newIncompleteFuture();
+        assertNotNull(future);
+        assertTrue(future instanceof PipelineControllerFuture);
+        assertNotSame(controller, future);
+        assertFalse(future.isDone());
     }
 
     @Test
@@ -208,22 +359,81 @@ public class PipelineControllerFutureTest {
         verify(future2).cancel(anyBoolean());
     }
 
+    /**
+     * Tests both wrap() methods.
+     */
     @Test
-    public void testAddFunction() {
-        AtomicReference<String> value = new AtomicReference<>();
+    public void testWrap() throws Exception {
+        controller = spy(controller);
 
-        Function<String, CompletableFuture<String>> func = controller.add(input -> {
-            value.set(input);
+        CompletableFuture<String> future = controller.wrap(compFuture);
+        verify(controller, never()).remove(compFuture);
+
+        compFuture.complete(TEXT);
+        assertEquals(TEXT, future.get());
+
+        verify(controller).remove(compFuture);
+    }
+
+    /**
+     * Tests wrap(), when the controller is not running.
+     */
+    @Test
+    public void testWrapNotRunning() throws Exception {
+        controller.cancel(false);
+        controller = spy(controller);
+
+        assertFalse(controller.wrap(compFuture).isDone());
+        verify(controller, never()).add(compFuture);
+        verify(controller, never()).remove(compFuture);
+
+        verify(compFuture).cancel(anyBoolean());
+    }
+
+    /**
+     * Tests wrap(), when the future throws an exception.
+     */
+    @Test
+    public void testWrapException() throws Exception {
+        controller = spy(controller);
+
+        CompletableFuture<String> future = controller.wrap(compFuture);
+        verify(controller, never()).remove(compFuture);
+
+        compFuture.completeExceptionally(EXPECTED_EXCEPTION);
+        assertThatThrownBy(() -> future.get()).hasCause(EXPECTED_EXCEPTION);
+
+        verify(controller).remove(compFuture);
+    }
+
+    @Test
+    public void testWrapFunction() throws Exception {
+
+        Function<String, CompletableFuture<String>> func = controller.wrap(input -> {
+            compFuture.complete(input);
             return compFuture;
         });
 
-        assertSame(compFuture, func.apply(TEXT));
-        assertEquals(TEXT, value.get());
+        CompletableFuture<String> future = func.apply(TEXT);
+        assertTrue(compFuture.isDone());
 
-        verify(compFuture, never()).cancel(anyBoolean());
+        assertEquals(TEXT, future.get());
 
         // should not have completed the controller
         assertFalse(controller.isDone());
+    }
+
+    /**
+     * Tests add(Function) when the controller is canceled after the future is added.
+     */
+    @Test
+    public void testWrapFunctionCancel() throws Exception {
+        Function<String, CompletableFuture<String>> func = controller.wrap(input -> compFuture);
+
+        CompletableFuture<String> future = func.apply(TEXT);
+        assertFalse(future.isDone());
+
+        assertFalse(compFuture.isDone());
 
         // cancel - should propagate
         controller.cancel(false);
@@ -235,10 +445,10 @@ public class PipelineControllerFutureTest {
      * Tests add(Function) when the controller is not running.
      */
     @Test
-    public void testAddFunctionNotRunning() {
+    public void testWrapFunctionNotRunning() {
         AtomicReference<String> value = new AtomicReference<>();
 
-        Function<String, CompletableFuture<String>> func = controller.add(input -> {
+        Function<String, CompletableFuture<String>> func = controller.wrap(input -> {
             value.set(input);
             return compFuture;
         });
@@ -251,4 +461,11 @@ public class PipelineControllerFutureTest {
 
         assertNull(value.get());
     }
+
+    private void verifyStopped() {
+        verify(runnable1).run();
+        verify(runnable2).run();
+        verify(future1).cancel(anyBoolean());
+        verify(future2).cancel(anyBoolean());
+    }
 }
diff --git a/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml b/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml
index f8a1e5112..c7fe46e47 100644
--- a/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml
+++ b/models-interactions/model-actors/actorServiceProvider/src/test/resources/logback-test.xml
@@ -35,8 +35,8 @@
         <appender-ref ref="STDOUT" />
     </root>
 
-    <!-- this is just an example -->
-    <logger name="ch.qos.logback.classic" level="off" additivity="false">
+    <!-- this is required for UtilTest -->
+    <logger name="org.onap.policy.controlloop.actorserviceprovider.Util" level="info" additivity="false">
         <appender-ref ref="STDOUT" />
     </logger>
 </configuration>