models-interactions/model-actors/actorServiceProvider/src/test/java/org/onap/policy/controlloop/actorserviceprovider/impl/OperationPartialTest.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP
   4  * ================================================================================
   5  * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.controlloop.actorserviceprovider.impl;
  22
  23 import static org.assertj.core.api.Assertions.assertThat;
  24 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
  25 import static org.junit.Assert.assertEquals;
  26 import static org.junit.Assert.assertFalse;
  27 import static org.junit.Assert.assertNotNull;
  28 import static org.junit.Assert.assertNull;
  29 import static org.junit.Assert.assertSame;
  30 import static org.junit.Assert.assertTrue;
  31 import static org.mockito.ArgumentMatchers.any;
  32 import static org.mockito.Mockito.verify;
  33 import static org.mockito.Mockito.when;
  34
  35 import ch.qos.logback.classic.Logger;
  36 import java.time.Instant;
  37 import java.util.ArrayDeque;
  38 import java.util.Arrays;
  39 import java.util.Deque;
  40 import java.util.LinkedList;
  41 import java.util.List;
  42 import java.util.Map;
  43 import java.util.Map.Entry;
  44 import java.util.UUID;
  45 import java.util.concurrent.CompletableFuture;
  46 import java.util.concurrent.CompletionException;
  47 import java.util.concurrent.ExecutionException;
  48 import java.util.concurrent.ForkJoinPool;
  49 import java.util.concurrent.Future;
  50 import java.util.concurrent.TimeUnit;
  51 import java.util.concurrent.TimeoutException;
  52 import java.util.concurrent.atomic.AtomicReference;
  53 import java.util.function.Consumer;
  54 import java.util.function.Supplier;
  55 import java.util.stream.Collectors;
  56 import lombok.Getter;
  57 import lombok.Setter;
  58 import org.junit.AfterClass;
  59 import org.junit.Before;
  60 import org.junit.BeforeClass;
  61 import org.junit.Test;
  62 import org.mockito.ArgumentCaptor;
  63 import org.mockito.Mock;
  64 import org.mockito.MockitoAnnotations;
  65 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
  66 import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
  67 import org.onap.policy.common.utils.coder.Coder;
  68 import org.onap.policy.common.utils.coder.CoderException;
  69 import org.onap.policy.common.utils.coder.StandardCoder;
  70 import org.onap.policy.common.utils.test.log.logback.ExtractAppender;
  71 import org.onap.policy.common.utils.time.PseudoExecutor;
  72 import org.onap.policy.controlloop.ControlLoopOperation;
  73 import org.onap.policy.controlloop.VirtualControlLoopEvent;
  74 import org.onap.policy.controlloop.actorserviceprovider.ActorService;
  75 import org.onap.policy.controlloop.actorserviceprovider.Operation;
  76 import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
  77 import org.onap.policy.controlloop.actorserviceprovider.Operator;
  78 import org.onap.policy.controlloop.actorserviceprovider.controlloop.ControlLoopEventContext;
  79 import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
  80 import org.onap.policy.controlloop.actorserviceprovider.parameters.OperatorConfig;
  81 import org.onap.policy.controlloop.actorserviceprovider.spi.Actor;
  82 import org.onap.policy.controlloop.policy.PolicyResult;
  83 import org.slf4j.LoggerFactory;
  84
  85 public class OperationPartialTest {
  86     private static final CommInfrastructure SINK_INFRA = CommInfrastructure.NOOP;
  87     private static final CommInfrastructure SOURCE_INFRA = CommInfrastructure.UEB;
  88     private static final int MAX_REQUESTS = 100;
  89     private static final int MAX_PARALLEL = 10;
  90     private static final String EXPECTED_EXCEPTION = "expected exception";
  91     private static final String ACTOR = "my-actor";
  92     private static final String OPERATION = "my-operation";
  93     private static final String MY_SINK = "my-sink";
  94     private static final String MY_SOURCE = "my-source";
  95     private static final String MY_TARGET_ENTITY = "my-entity";
  96     private static final String TEXT = "my-text";
  97     private static final int TIMEOUT = 1000;
  98     private static final UUID REQ_ID = UUID.randomUUID();
  99
 100     private static final List<PolicyResult> FAILURE_RESULTS = Arrays.asList(PolicyResult.values()).stream()
 101                     .filter(result -> result != PolicyResult.SUCCESS).collect(Collectors.toList());
 102
 103     /**
 104      * Used to attach an appender to the class' logger.
 105      */
 106     private static final Logger logger = (Logger) LoggerFactory.getLogger(OperationPartial.class);
 107     private static final ExtractAppender appender = new ExtractAppender();
 108
 109     @Mock
 110     private ActorService service;
 111     @Mock
 112     private Actor guardActor;
 113     @Mock
 114     private Operator guardOperator;
 115     @Mock
 116     private Operation guardOperation;
 117
 118     private VirtualControlLoopEvent event;
 119     private ControlLoopEventContext context;
 120     private PseudoExecutor executor;
 121     private ControlLoopOperationParams params;
 122
 123     private MyOper oper;
 124
 125     private int numStart;
 126     private int numEnd;
 127
 128     private Instant tstart;
 129
 130     private OperationOutcome opstart;
 131     private OperationOutcome opend;
 132
 133     private Deque<OperationOutcome> starts;
 134     private Deque<OperationOutcome> ends;
 135
 136     private OperatorConfig config;
 137
 138     /**
 139      * Attaches the appender to the logger.
 140      */
 141     @BeforeClass
 142     public static void setUpBeforeClass() throws Exception {
 143         /*
 144          * Attach appender to the logger.
 145          */
 146         appender.setContext(logger.getLoggerContext());
 147         appender.start();
 148
 149         logger.addAppender(appender);
 150     }
 151
 152     /**
 153      * Stops the appender.
 154      */
 155     @AfterClass
 156     public static void tearDownAfterClass() {
 157         appender.stop();
 158     }
 159
 160     /**
 161      * Initializes the fields, including {@link #oper}.
 162      */
 163     @Before
 164     public void setUp() {
 165         MockitoAnnotations.initMocks(this);
 166
 167         event = new VirtualControlLoopEvent();
 168         event.setRequestId(REQ_ID);
 169
 170         context = new ControlLoopEventContext(event);
 171         executor = new PseudoExecutor();
 172
 173         params = ControlLoopOperationParams.builder().completeCallback(this::completer).context(context)
 174                         .executor(executor).actorService(service).actor(ACTOR).operation(OPERATION).timeoutSec(TIMEOUT)
 175                         .startCallback(this::starter).targetEntity(MY_TARGET_ENTITY).build();
 176
 177         when(service.getActor(OperationPartial.GUARD_ACTOR_NAME)).thenReturn(guardActor);
 178         when(guardActor.getOperator(OperationPartial.GUARD_OPERATION_NAME)).thenReturn(guardOperator);
 179         when(guardOperator.buildOperation(any())).thenReturn(guardOperation);
 180         when(guardOperation.start()).thenReturn(CompletableFuture.completedFuture(makeSuccess()));
 181
 182         config = new OperatorConfig(executor);
 183
 184         oper = new MyOper();
 185
 186         tstart = null;
 187
 188         opstart = null;
 189         opend = null;
 190
 191         starts = new ArrayDeque<>(10);
 192         ends = new ArrayDeque<>(10);
 193     }
 194
 195     @Test
 196     public void testOperatorPartial_testGetActorName_testGetName() {
 197         assertEquals(ACTOR, oper.getActorName());
 198         assertEquals(OPERATION, oper.getName());
 199         assertEquals(ACTOR + "." + OPERATION, oper.getFullName());
 200     }
 201
 202     @Test
 203     public void testGetBlockingThread() throws Exception {
 204         CompletableFuture<Void> future = new CompletableFuture<>();
 205
 206         // use the real executor
 207         OperatorPartial oper2 = new OperatorPartial(ACTOR, OPERATION) {
 208             @Override
 209             public Operation buildOperation(ControlLoopOperationParams params) {
 210                 return null;
 211             }
 212         };
 213
 214         oper2.getBlockingExecutor().execute(() -> future.complete(null));
 215
 216         assertNull(future.get(5, TimeUnit.SECONDS));
 217     }
 218
 219     @Test
 220     public void testStart() {
 221         verifyRun("testStart", 1, 1, PolicyResult.SUCCESS);
 222     }
 223
 224     /**
 225      * Tests start() with multiple running requests.
 226      */
 227     @Test
 228     public void testStartMultiple() {
 229         for (int count = 0; count < MAX_PARALLEL; ++count) {
 230             oper.start();
 231         }
 232
 233         assertTrue(executor.runAll(MAX_REQUESTS * MAX_PARALLEL));
 234
 235         assertNotNull(opstart);
 236         assertNotNull(opend);
 237         assertEquals(PolicyResult.SUCCESS, opend.getResult());
 238
 239         assertEquals(MAX_PARALLEL, numStart);
 240         assertEquals(MAX_PARALLEL, oper.getCount());
 241         assertEquals(MAX_PARALLEL, numEnd);
 242     }
 243
 244     /**
 245      * Tests startPreprocessor() when the preprocessor returns a failure.
 246      */
 247     @Test
 248     public void testStartPreprocessorFailure() {
 249         oper.setPreProc(CompletableFuture.completedFuture(makeFailure()));
 250
 251         verifyRun("testStartPreprocessorFailure", 1, 0, PolicyResult.FAILURE_GUARD);
 252     }
 253
 254     /**
 255      * Tests startPreprocessor() when the preprocessor throws an exception.
 256      */
 257     @Test
 258     public void testStartPreprocessorException() {
 259         // arrange for the preprocessor to throw an exception
 260         oper.setPreProc(CompletableFuture.failedFuture(new IllegalStateException(EXPECTED_EXCEPTION)));
 261
 262         verifyRun("testStartPreprocessorException", 1, 0, PolicyResult.FAILURE_GUARD);
 263     }
 264
 265     /**
 266      * Tests startPreprocessor() when the pipeline is not running.
 267      */
 268     @Test
 269     public void testStartPreprocessorNotRunning() {
 270         // arrange for the preprocessor to return success, which will be ignored
 271         // oper.setGuard(CompletableFuture.completedFuture(makeSuccess()));
 272
 273         oper.start().cancel(false);
 274         assertTrue(executor.runAll(MAX_REQUESTS));
 275
 276         assertNull(opstart);
 277         assertNull(opend);
 278
 279         assertEquals(0, numStart);
 280         assertEquals(0, oper.getCount());
 281         assertEquals(0, numEnd);
 282     }
 283
 284     /**
 285      * Tests startPreprocessor() when the preprocessor <b>builder</b> throws an exception.
 286      */
 287     @Test
 288     public void testStartPreprocessorBuilderException() {
 289         oper = new MyOper() {
 290             @Override
 291             protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
 292                 throw new IllegalStateException(EXPECTED_EXCEPTION);
 293             }
 294         };
 295
 296         assertThatIllegalStateException().isThrownBy(() -> oper.start());
 297
 298         // should be nothing in the queue
 299         assertEquals(0, executor.getQueueLength());
 300     }
 301
 302     @Test
 303     public void testStartPreprocessorAsync() {
 304         assertNull(oper.startPreprocessorAsync());
 305     }
 306
 307     @Test
 308     public void testStartGuardAsync() throws Exception {
 309         CompletableFuture<OperationOutcome> future = oper.startGuardAsync();
 310         assertTrue(future.isDone());
 311         assertEquals(PolicyResult.SUCCESS, future.get().getResult());
 312
 313         // verify the parameters that were passed
 314         ArgumentCaptor<ControlLoopOperationParams> paramsCaptor =
 315                         ArgumentCaptor.forClass(ControlLoopOperationParams.class);
 316         verify(guardOperator).buildOperation(paramsCaptor.capture());
 317
 318         params = paramsCaptor.getValue();
 319         assertEquals(OperationPartial.GUARD_ACTOR_NAME, params.getActor());
 320         assertEquals(OperationPartial.GUARD_OPERATION_NAME, params.getOperation());
 321         assertNull(params.getRetry());
 322         assertNull(params.getTimeoutSec());
 323
 324         Map<String, Object> payload = params.getPayload();
 325         assertNotNull(payload);
 326
 327         assertEquals(oper.makeGuardPayload(), payload);
 328     }
 329
 330     @Test
 331     public void testMakeGuardPayload() {
 332         Map<String, Object> payload = oper.makeGuardPayload();
 333         assertSame(REQ_ID, payload.get("requestId"));
 334
 335         // request id changes, so remove it
 336         payload.remove("requestId");
 337
 338         assertEquals("{actor=my-actor, operation=my-operation, target=my-entity}", payload.toString());
 339
 340         // repeat, but with closed loop name
 341         event.setClosedLoopControlName("my-loop");
 342         payload = oper.makeGuardPayload();
 343         payload.remove("requestId");
 344         assertEquals("{actor=my-actor, operation=my-operation, target=my-entity, clname=my-loop}", payload.toString());
 345     }
 346
 347     @Test
 348     public void testStartOperationAsync() {
 349         oper.start();
 350         assertTrue(executor.runAll(MAX_REQUESTS));
 351
 352         assertEquals(1, oper.getCount());
 353     }
 354
 355     @Test
 356     public void testIsSuccess() {
 357         assertFalse(oper.isSuccess(null));
 358
 359         OperationOutcome outcome = new OperationOutcome();
 360
 361         outcome.setResult(PolicyResult.SUCCESS);
 362         assertTrue(oper.isSuccess(outcome));
 363
 364         for (PolicyResult failure : FAILURE_RESULTS) {
 365             outcome.setResult(failure);
 366             assertFalse("testIsSuccess-" + failure, oper.isSuccess(outcome));
 367         }
 368     }
 369
 370     @Test
 371     public void testIsActorFailed() {
 372         assertFalse(oper.isActorFailed(null));
 373
 374         OperationOutcome outcome = params.makeOutcome();
 375
 376         // incorrect outcome
 377         outcome.setResult(PolicyResult.SUCCESS);
 378         assertFalse(oper.isActorFailed(outcome));
 379
 380         outcome.setResult(PolicyResult.FAILURE_RETRIES);
 381         assertFalse(oper.isActorFailed(outcome));
 382
 383         // correct outcome
 384         outcome.setResult(PolicyResult.FAILURE);
 385
 386         // incorrect actor
 387         outcome.setActor(MY_SINK);
 388         assertFalse(oper.isActorFailed(outcome));
 389         outcome.setActor(null);
 390         assertFalse(oper.isActorFailed(outcome));
 391         outcome.setActor(ACTOR);
 392
 393         // incorrect operation
 394         outcome.setOperation(MY_SINK);
 395         assertFalse(oper.isActorFailed(outcome));
 396         outcome.setOperation(null);
 397         assertFalse(oper.isActorFailed(outcome));
 398         outcome.setOperation(OPERATION);
 399
 400         // correct values
 401         assertTrue(oper.isActorFailed(outcome));
 402     }
 403
 404     @Test
 405     public void testDoOperation() {
 406         /*
 407          * Use an operation that doesn't override doOperation().
 408          */
 409         OperationPartial oper2 = new OperationPartial(params, config) {};
 410
 411         oper2.start();
 412         assertTrue(executor.runAll(MAX_REQUESTS));
 413
 414         assertNotNull(opend);
 415         assertEquals(PolicyResult.FAILURE_EXCEPTION, opend.getResult());
 416     }
 417
 418     @Test
 419     public void testTimeout() throws Exception {
 420
 421         // use a real executor
 422         params = params.toBuilder().executor(ForkJoinPool.commonPool()).build();
 423
 424         // trigger timeout very quickly
 425         oper = new MyOper() {
 426             @Override
 427             protected long getTimeoutMs(Integer timeoutSec) {
 428                 return 1;
 429             }
 430
 431             @Override
 432             protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
 433
 434                 OperationOutcome outcome2 = params.makeOutcome();
 435                 outcome2.setResult(PolicyResult.SUCCESS);
 436
 437                 /*
 438                  * Create an incomplete future that will timeout after the operation's
 439                  * timeout. If it fires before the other timer, then it will return a
 440                  * SUCCESS outcome.
 441                  */
 442                 CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
 443                 future = future.orTimeout(1, TimeUnit.SECONDS).handleAsync((unused1, unused2) -> outcome,
 444                                 params.getExecutor());
 445
 446                 return future;
 447             }
 448         };
 449
 450         assertEquals(PolicyResult.FAILURE_TIMEOUT, oper.start().get().getResult());
 451     }
 452
 453     /**
 454      * Tests retry functions, when the count is set to zero and retries are exhausted.
 455      */
 456     @Test
 457     public void testSetRetryFlag_testRetryOnFailure_ZeroRetries_testStartOperationAttempt() {
 458         params = params.toBuilder().retry(0).build();
 459
 460         // new params, thus need a new operation
 461         oper = new MyOper();
 462
 463         oper.setMaxFailures(10);
 464
 465         verifyRun("testSetRetryFlag_testRetryOnFailure_ZeroRetries", 1, 1, PolicyResult.FAILURE);
 466     }
 467
 468     /**
 469      * Tests retry functions, when the count is null and retries are exhausted.
 470      */
 471     @Test
 472     public void testSetRetryFlag_testRetryOnFailure_NullRetries() {
 473         params = params.toBuilder().retry(null).build();
 474
 475         // new params, thus need a new operation
 476         oper = new MyOper();
 477
 478         oper.setMaxFailures(10);
 479
 480         verifyRun("testSetRetryFlag_testRetryOnFailure_NullRetries", 1, 1, PolicyResult.FAILURE);
 481     }
 482
 483     /**
 484      * Tests retry functions, when retries are exhausted.
 485      */
 486     @Test
 487     public void testSetRetryFlag_testRetryOnFailure_RetriesExhausted() {
 488         final int maxRetries = 3;
 489         params = params.toBuilder().retry(maxRetries).build();
 490
 491         // new params, thus need a new operation
 492         oper = new MyOper();
 493
 494         oper.setMaxFailures(10);
 495
 496         verifyRun("testSetRetryFlag_testRetryOnFailure_RetriesExhausted", maxRetries + 1, maxRetries + 1,
 497                         PolicyResult.FAILURE_RETRIES);
 498     }
 499
 500     /**
 501      * Tests retry functions, when a success follows some retries.
 502      */
 503     @Test
 504     public void testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries() {
 505         params = params.toBuilder().retry(10).build();
 506
 507         // new params, thus need a new operation
 508         oper = new MyOper();
 509
 510         final int maxFailures = 3;
 511         oper.setMaxFailures(maxFailures);
 512
 513         verifyRun("testSetRetryFlag_testRetryOnFailure_SuccessAfterRetries", maxFailures + 1, maxFailures + 1,
 514                         PolicyResult.SUCCESS);
 515     }
 516
 517     /**
 518      * Tests retry functions, when the outcome is {@code null}.
 519      */
 520     @Test
 521     public void testSetRetryFlag_testRetryOnFailure_NullOutcome() {
 522
 523         // arrange to return null from doOperation()
 524         oper = new MyOper() {
 525             @Override
 526             protected OperationOutcome doOperation(int attempt, OperationOutcome outcome) {
 527
 528                 // update counters
 529                 super.doOperation(attempt, outcome);
 530                 return null;
 531             }
 532         };
 533
 534         verifyRun("testSetRetryFlag_testRetryOnFailure_NullOutcome", 1, 1, PolicyResult.FAILURE, noop());
 535     }
 536
 537     @Test
 538     public void testSleep() throws Exception {
 539         CompletableFuture<Void> future = oper.sleep(-1, TimeUnit.SECONDS);
 540         assertTrue(future.isDone());
 541         assertNull(future.get());
 542
 543         // edge case
 544         future = oper.sleep(0, TimeUnit.SECONDS);
 545         assertTrue(future.isDone());
 546         assertNull(future.get());
 547
 548         /*
 549          * Start a second sleep we can use to check the first while it's running.
 550          */
 551         tstart = Instant.now();
 552         future = oper.sleep(100, TimeUnit.MILLISECONDS);
 553
 554         CompletableFuture<Void> future2 = oper.sleep(10, TimeUnit.MILLISECONDS);
 555
 556         // wait for second to complete and verify that the first has not completed
 557         future2.get();
 558         assertFalse(future.isDone());
 559
 560         // wait for second to complete
 561         future.get();
 562
 563         long diff = Instant.now().toEpochMilli() - tstart.toEpochMilli();
 564         assertTrue(diff >= 99);
 565     }
 566
 567     @Test
 568     public void testIsSameOperation() {
 569         assertFalse(oper.isSameOperation(null));
 570
 571         OperationOutcome outcome = params.makeOutcome();
 572
 573         // wrong actor - should be false
 574         outcome.setActor(null);
 575         assertFalse(oper.isSameOperation(outcome));
 576         outcome.setActor(MY_SINK);
 577         assertFalse(oper.isSameOperation(outcome));
 578         outcome.setActor(ACTOR);
 579
 580         // wrong operation - should be null
 581         outcome.setOperation(null);
 582         assertFalse(oper.isSameOperation(outcome));
 583         outcome.setOperation(MY_SINK);
 584         assertFalse(oper.isSameOperation(outcome));
 585         outcome.setOperation(OPERATION);
 586
 587         assertTrue(oper.isSameOperation(outcome));
 588     }
 589
 590     /**
 591      * Tests handleFailure() when the outcome is a success.
 592      */
 593     @Test
 594     public void testHandlePreprocessorFailureSuccess() {
 595         oper.setPreProc(CompletableFuture.completedFuture(makeSuccess()));
 596         verifyRun("testHandlePreprocessorFailureTrue", 1, 1, PolicyResult.SUCCESS);
 597     }
 598
 599     /**
 600      * Tests handleFailure() when the outcome is <i>not</i> a success.
 601      */
 602     @Test
 603     public void testHandlePreprocessorFailureFailed() throws Exception {
 604         oper.setPreProc(CompletableFuture.completedFuture(makeFailure()));
 605         verifyRun("testHandlePreprocessorFailureFalse", 1, 0, PolicyResult.FAILURE_GUARD);
 606     }
 607
 608     /**
 609      * Tests handleFailure() when the outcome is {@code null}.
 610      */
 611     @Test
 612     public void testHandlePreprocessorFailureNull() throws Exception {
 613         // arrange to return a null outcome from the preprocessor
 614         oper.setPreProc(CompletableFuture.completedFuture(null));
 615         verifyRun("testHandlePreprocessorFailureNull", 1, 0, PolicyResult.FAILURE_GUARD);
 616     }
 617
 618     @Test
 619     public void testFromException() {
 620         // arrange to generate an exception when operation runs
 621         oper.setGenException(true);
 622
 623         verifyRun("testFromException", 1, 1, PolicyResult.FAILURE_EXCEPTION);
 624     }
 625
 626     /**
 627      * Tests fromException() when there is no exception.
 628      */
 629     @Test
 630     public void testFromExceptionNoExcept() {
 631         verifyRun("testFromExceptionNoExcept", 1, 1, PolicyResult.SUCCESS);
 632     }
 633
 634     /**
 635      * Tests both flavors of anyOf(), because one invokes the other.
 636      */
 637     @Test
 638     public void testAnyOf() throws Exception {
 639         // first task completes, others do not
 640         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 641
 642         final OperationOutcome outcome = params.makeOutcome();
 643
 644         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 645         tasks.add(() -> new CompletableFuture<>());
 646         tasks.add(() -> null);
 647         tasks.add(() -> new CompletableFuture<>());
 648
 649         CompletableFuture<OperationOutcome> result = oper.anyOf(tasks);
 650         assertTrue(executor.runAll(MAX_REQUESTS));
 651         assertTrue(result.isDone());
 652         assertSame(outcome, result.get());
 653
 654         // repeat using array form
 655         @SuppressWarnings("unchecked")
 656         Supplier<CompletableFuture<OperationOutcome>>[] taskArray = new Supplier[tasks.size()];
 657         result = oper.anyOf(tasks.toArray(taskArray));
 658         assertTrue(executor.runAll(MAX_REQUESTS));
 659         assertTrue(result.isDone());
 660         assertSame(outcome, result.get());
 661
 662         // second task completes, others do not
 663         tasks.clear();
 664         tasks.add(() -> new CompletableFuture<>());
 665         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 666         tasks.add(() -> new CompletableFuture<>());
 667
 668         result = oper.anyOf(tasks);
 669         assertTrue(executor.runAll(MAX_REQUESTS));
 670         assertTrue(result.isDone());
 671         assertSame(outcome, result.get());
 672
 673         // third task completes, others do not
 674         tasks.clear();
 675         tasks.add(() -> new CompletableFuture<>());
 676         tasks.add(() -> new CompletableFuture<>());
 677         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 678
 679         result = oper.anyOf(tasks);
 680         assertTrue(executor.runAll(MAX_REQUESTS));
 681         assertTrue(result.isDone());
 682         assertSame(outcome, result.get());
 683     }
 684
 685     /**
 686      * Tests both flavors of anyOf(), for edge cases: zero items, and one item.
 687      */
 688     @Test
 689     @SuppressWarnings("unchecked")
 690     public void testAnyOfEdge() throws Exception {
 691         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 692
 693         // zero items: check both using a list and using an array
 694         assertNull(oper.anyOf(tasks));
 695         assertNull(oper.anyOf());
 696
 697         // one item: : check both using a list and using an array
 698         CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
 699         tasks.add(() -> future1);
 700
 701         assertSame(future1, oper.anyOf(tasks));
 702         assertSame(future1, oper.anyOf(() -> future1));
 703     }
 704
 705     @Test
 706     public void testAllOfArray() throws Exception {
 707         final OperationOutcome outcome = params.makeOutcome();
 708
 709         CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
 710         CompletableFuture<OperationOutcome> future2 = new CompletableFuture<>();
 711         CompletableFuture<OperationOutcome> future3 = new CompletableFuture<>();
 712
 713         @SuppressWarnings("unchecked")
 714         CompletableFuture<OperationOutcome> result =
 715                         oper.allOf(() -> future1, () -> future2, () -> null, () -> future3);
 716
 717         assertTrue(executor.runAll(MAX_REQUESTS));
 718         assertFalse(result.isDone());
 719         future1.complete(outcome);
 720
 721         // complete 3 before 2
 722         assertTrue(executor.runAll(MAX_REQUESTS));
 723         assertFalse(result.isDone());
 724         future3.complete(outcome);
 725
 726         assertTrue(executor.runAll(MAX_REQUESTS));
 727         assertFalse(result.isDone());
 728         future2.complete(outcome);
 729
 730         // all of them are now done
 731         assertTrue(executor.runAll(MAX_REQUESTS));
 732         assertTrue(result.isDone());
 733         assertSame(outcome, result.get());
 734     }
 735
 736     @Test
 737     public void testAllOfList() throws Exception {
 738         final OperationOutcome outcome = params.makeOutcome();
 739
 740         CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
 741         CompletableFuture<OperationOutcome> future2 = new CompletableFuture<>();
 742         CompletableFuture<OperationOutcome> future3 = new CompletableFuture<>();
 743
 744         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 745         tasks.add(() -> future1);
 746         tasks.add(() -> future2);
 747         tasks.add(() -> null);
 748         tasks.add(() -> future3);
 749
 750         CompletableFuture<OperationOutcome> result = oper.allOf(tasks);
 751
 752         assertTrue(executor.runAll(MAX_REQUESTS));
 753         assertFalse(result.isDone());
 754         future1.complete(outcome);
 755
 756         // complete 3 before 2
 757         assertTrue(executor.runAll(MAX_REQUESTS));
 758         assertFalse(result.isDone());
 759         future3.complete(outcome);
 760
 761         assertTrue(executor.runAll(MAX_REQUESTS));
 762         assertFalse(result.isDone());
 763         future2.complete(outcome);
 764
 765         // all of them are now done
 766         assertTrue(executor.runAll(MAX_REQUESTS));
 767         assertTrue(result.isDone());
 768         assertSame(outcome, result.get());
 769     }
 770
 771     /**
 772      * Tests both flavors of allOf(), for edge cases: zero items, and one item.
 773      */
 774     @Test
 775     @SuppressWarnings("unchecked")
 776     public void testAllOfEdge() throws Exception {
 777         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 778
 779         // zero items: check both using a list and using an array
 780         assertNull(oper.allOf(tasks));
 781         assertNull(oper.allOf());
 782
 783         // one item: : check both using a list and using an array
 784         CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
 785         tasks.add(() -> future1);
 786
 787         assertSame(future1, oper.allOf(tasks));
 788         assertSame(future1, oper.allOf(() -> future1));
 789     }
 790
 791     @Test
 792     public void testAttachFutures() throws Exception {
 793         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 794
 795         // third task throws an exception during construction
 796         CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
 797         CompletableFuture<OperationOutcome> future2 = new CompletableFuture<>();
 798         CompletableFuture<OperationOutcome> future3 = new CompletableFuture<>();
 799         tasks.add(() -> future1);
 800         tasks.add(() -> future2);
 801         tasks.add(() -> {
 802             throw new IllegalStateException(EXPECTED_EXCEPTION);
 803         });
 804         tasks.add(() -> future3);
 805
 806         assertThatIllegalStateException().isThrownBy(() -> oper.anyOf(tasks)).withMessage(EXPECTED_EXCEPTION);
 807
 808         // should have canceled the first two, but not the last
 809         assertTrue(future1.isCancelled());
 810         assertTrue(future2.isCancelled());
 811         assertFalse(future3.isCancelled());
 812     }
 813
 814     @Test
 815     public void testCombineOutcomes() throws Exception {
 816         // only one outcome
 817         verifyOutcomes(0, PolicyResult.SUCCESS);
 818         verifyOutcomes(0, PolicyResult.FAILURE_EXCEPTION);
 819
 820         // maximum is in different positions
 821         verifyOutcomes(0, PolicyResult.FAILURE, PolicyResult.SUCCESS, PolicyResult.FAILURE_GUARD);
 822         verifyOutcomes(1, PolicyResult.SUCCESS, PolicyResult.FAILURE, PolicyResult.FAILURE_GUARD);
 823         verifyOutcomes(2, PolicyResult.SUCCESS, PolicyResult.FAILURE_GUARD, PolicyResult.FAILURE);
 824
 825         // null outcome - takes precedence over a success
 826         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 827         tasks.add(() -> CompletableFuture.completedFuture(params.makeOutcome()));
 828         tasks.add(() -> CompletableFuture.completedFuture(null));
 829         tasks.add(() -> CompletableFuture.completedFuture(params.makeOutcome()));
 830         CompletableFuture<OperationOutcome> result = oper.allOf(tasks);
 831
 832         assertTrue(executor.runAll(MAX_REQUESTS));
 833         assertTrue(result.isDone());
 834         assertNull(result.get());
 835
 836         // one throws an exception during execution
 837         IllegalStateException except = new IllegalStateException(EXPECTED_EXCEPTION);
 838
 839         tasks.clear();
 840         tasks.add(() -> CompletableFuture.completedFuture(params.makeOutcome()));
 841         tasks.add(() -> CompletableFuture.failedFuture(except));
 842         tasks.add(() -> CompletableFuture.completedFuture(params.makeOutcome()));
 843         result = oper.allOf(tasks);
 844
 845         assertTrue(executor.runAll(MAX_REQUESTS));
 846         assertTrue(result.isCompletedExceptionally());
 847         result.whenComplete((unused, thrown) -> assertSame(except, thrown));
 848     }
 849
 850     /**
 851      * Tests both flavors of sequence(), because one invokes the other.
 852      */
 853     @Test
 854     public void testSequence() throws Exception {
 855         final OperationOutcome outcome = params.makeOutcome();
 856
 857         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 858         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 859         tasks.add(() -> null);
 860         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 861         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 862
 863         CompletableFuture<OperationOutcome> result = oper.sequence(tasks);
 864         assertTrue(executor.runAll(MAX_REQUESTS));
 865         assertTrue(result.isDone());
 866         assertSame(outcome, result.get());
 867
 868         // repeat using array form
 869         @SuppressWarnings("unchecked")
 870         Supplier<CompletableFuture<OperationOutcome>>[] taskArray = new Supplier[tasks.size()];
 871         result = oper.sequence(tasks.toArray(taskArray));
 872         assertTrue(executor.runAll(MAX_REQUESTS));
 873         assertTrue(result.isDone());
 874         assertSame(outcome, result.get());
 875
 876         // second task fails, third should not run
 877         OperationOutcome failure = params.makeOutcome();
 878         failure.setResult(PolicyResult.FAILURE);
 879         tasks.clear();
 880         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 881         tasks.add(() -> CompletableFuture.completedFuture(failure));
 882         tasks.add(() -> CompletableFuture.completedFuture(outcome));
 883
 884         result = oper.sequence(tasks);
 885         assertTrue(executor.runAll(MAX_REQUESTS));
 886         assertTrue(result.isDone());
 887         assertSame(failure, result.get());
 888     }
 889
 890     /**
 891      * Tests both flavors of sequence(), for edge cases: zero items, and one item.
 892      */
 893     @Test
 894     @SuppressWarnings("unchecked")
 895     public void testSequenceEdge() throws Exception {
 896         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 897
 898         // zero items: check both using a list and using an array
 899         assertNull(oper.sequence(tasks));
 900         assertNull(oper.sequence());
 901
 902         // one item: : check both using a list and using an array
 903         CompletableFuture<OperationOutcome> future1 = new CompletableFuture<>();
 904         tasks.add(() -> future1);
 905
 906         assertSame(future1, oper.sequence(tasks));
 907         assertSame(future1, oper.sequence(() -> future1));
 908     }
 909
 910     private void verifyOutcomes(int expected, PolicyResult... results) throws Exception {
 911         List<Supplier<CompletableFuture<OperationOutcome>>> tasks = new LinkedList<>();
 912
 913         OperationOutcome expectedOutcome = null;
 914
 915         for (int count = 0; count < results.length; ++count) {
 916             OperationOutcome outcome = params.makeOutcome();
 917             outcome.setResult(results[count]);
 918             tasks.add(() -> CompletableFuture.completedFuture(outcome));
 919
 920             if (count == expected) {
 921                 expectedOutcome = outcome;
 922             }
 923         }
 924
 925         CompletableFuture<OperationOutcome> result = oper.allOf(tasks);
 926
 927         assertTrue(executor.runAll(MAX_REQUESTS));
 928         assertTrue(result.isDone());
 929         assertSame(expectedOutcome, result.get());
 930     }
 931
 932     @Test
 933     public void testDetmPriority() throws CoderException {
 934         assertEquals(1, oper.detmPriority(null));
 935
 936         OperationOutcome outcome = params.makeOutcome();
 937
 938         Map<PolicyResult, Integer> map = Map.of(PolicyResult.SUCCESS, 0, PolicyResult.FAILURE_GUARD, 2,
 939                         PolicyResult.FAILURE_RETRIES, 3, PolicyResult.FAILURE, 4, PolicyResult.FAILURE_TIMEOUT, 5,
 940                         PolicyResult.FAILURE_EXCEPTION, 6);
 941
 942         for (Entry<PolicyResult, Integer> ent : map.entrySet()) {
 943             outcome.setResult(ent.getKey());
 944             assertEquals(ent.getKey().toString(), ent.getValue().intValue(), oper.detmPriority(outcome));
 945         }
 946
 947         /*
 948          * Test null result. We can't actually set it to null, because the set() method
 949          * won't allow it. Instead, we decode it from a structure.
 950          */
 951         outcome = new StandardCoder().decode("{\"result\":null}", OperationOutcome.class);
 952         assertEquals(1, oper.detmPriority(outcome));
 953     }
 954
 955     /**
 956      * Tests callbackStarted() when the pipeline has already been stopped.
 957      */
 958     @Test
 959     public void testCallbackStartedNotRunning() {
 960         AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
 961
 962         /*
 963          * arrange to stop the controller when the start-callback is invoked, but capture
 964          * the outcome
 965          */
 966         params = params.toBuilder().startCallback(oper -> {
 967             starter(oper);
 968             future.get().cancel(false);
 969         }).build();
 970
 971         // new params, thus need a new operation
 972         oper = new MyOper();
 973
 974         future.set(oper.start());
 975         assertTrue(executor.runAll(MAX_REQUESTS));
 976
 977         // should have only run once
 978         assertEquals(1, numStart);
 979     }
 980
 981     /**
 982      * Tests callbackCompleted() when the pipeline has already been stopped.
 983      */
 984     @Test
 985     public void testCallbackCompletedNotRunning() {
 986         AtomicReference<Future<OperationOutcome>> future = new AtomicReference<>();
 987
 988         // arrange to stop the controller when the start-callback is invoked
 989         params = params.toBuilder().startCallback(oper -> {
 990             future.get().cancel(false);
 991         }).build();
 992
 993         // new params, thus need a new operation
 994         oper = new MyOper();
 995
 996         future.set(oper.start());
 997         assertTrue(executor.runAll(MAX_REQUESTS));
 998
 999         // should not have been set
1000         assertNull(opend);
1001         assertEquals(0, numEnd);
1002     }
1003
1004     @Test
1005     public void testSetOutcomeControlLoopOperationOutcomeThrowable() {
1006         final CompletionException timex = new CompletionException(new TimeoutException(EXPECTED_EXCEPTION));
1007
1008         OperationOutcome outcome;
1009
1010         outcome = new OperationOutcome();
1011         oper.setOutcome(outcome, timex);
1012         assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
1013         assertEquals(PolicyResult.FAILURE_TIMEOUT, outcome.getResult());
1014
1015         outcome = new OperationOutcome();
1016         oper.setOutcome(outcome, new IllegalStateException(EXPECTED_EXCEPTION));
1017         assertEquals(ControlLoopOperation.FAILED_MSG, outcome.getMessage());
1018         assertEquals(PolicyResult.FAILURE_EXCEPTION, outcome.getResult());
1019     }
1020
1021     @Test
1022     public void testSetOutcomeControlLoopOperationOutcomePolicyResult() {
1023         OperationOutcome outcome;
1024
1025         outcome = new OperationOutcome();
1026         oper.setOutcome(outcome, PolicyResult.SUCCESS);
1027         assertEquals(ControlLoopOperation.SUCCESS_MSG, outcome.getMessage());
1028         assertEquals(PolicyResult.SUCCESS, outcome.getResult());
1029
1030         oper.setOutcome(outcome, PolicyResult.SUCCESS);
1031         assertEquals(ControlLoopOperation.SUCCESS_MSG, outcome.getMessage());
1032         assertEquals(PolicyResult.SUCCESS, outcome.getResult());
1033
1034         for (PolicyResult result : FAILURE_RESULTS) {
1035             outcome = new OperationOutcome();
1036             oper.setOutcome(outcome, result);
1037             assertEquals(result.toString(), ControlLoopOperation.FAILED_MSG, outcome.getMessage());
1038             assertEquals(result.toString(), result, outcome.getResult());
1039         }
1040     }
1041
1042     @Test
1043     public void testIsTimeout() {
1044         final TimeoutException timex = new TimeoutException(EXPECTED_EXCEPTION);
1045
1046         assertFalse(oper.isTimeout(new IllegalStateException(EXPECTED_EXCEPTION)));
1047         assertFalse(oper.isTimeout(new IllegalStateException(timex)));
1048         assertFalse(oper.isTimeout(new CompletionException(new IllegalStateException(timex))));
1049         assertFalse(oper.isTimeout(new CompletionException(null)));
1050         assertFalse(oper.isTimeout(new CompletionException(new CompletionException(timex))));
1051
1052         assertTrue(oper.isTimeout(timex));
1053         assertTrue(oper.isTimeout(new CompletionException(timex)));
1054     }
1055
1056     @Test
1057     public void testLogMessage() {
1058         final String infraStr = SINK_INFRA.toString();
1059
1060         // log structured data
1061         appender.clearExtractions();
1062         oper.logMessage(EventType.OUT, SINK_INFRA, MY_SINK, new MyData());
1063         List<String> output = appender.getExtracted();
1064         assertEquals(1, output.size());
1065
1066         assertThat(output.get(0)).contains(infraStr).contains(MY_SINK).contains("OUT")
1067                         .contains("{\n  \"text\": \"my-text\"\n}");
1068
1069         // repeat with a response
1070         appender.clearExtractions();
1071         oper.logMessage(EventType.IN, SOURCE_INFRA, MY_SOURCE, new MyData());
1072         output = appender.getExtracted();
1073         assertEquals(1, output.size());
1074
1075         assertThat(output.get(0)).contains(SOURCE_INFRA.toString()).contains(MY_SOURCE).contains("IN")
1076                         .contains("{\n  \"text\": \"my-text\"\n}");
1077
1078         // log a plain string
1079         appender.clearExtractions();
1080         oper.logMessage(EventType.OUT, SINK_INFRA, MY_SINK, TEXT);
1081         output = appender.getExtracted();
1082         assertEquals(1, output.size());
1083         assertThat(output.get(0)).contains(infraStr).contains(MY_SINK).contains(TEXT);
1084
1085         // log a null request
1086         appender.clearExtractions();
1087         oper.logMessage(EventType.OUT, SINK_INFRA, MY_SINK, null);
1088         output = appender.getExtracted();
1089         assertEquals(1, output.size());
1090
1091         assertThat(output.get(0)).contains(infraStr).contains(MY_SINK).contains("null");
1092
1093         // generate exception from coder
1094         setOperCoderException();
1095
1096         appender.clearExtractions();
1097         oper.logMessage(EventType.OUT, SINK_INFRA, MY_SINK, new MyData());
1098         output = appender.getExtracted();
1099         assertEquals(2, output.size());
1100         assertThat(output.get(0)).contains("cannot pretty-print request");
1101         assertThat(output.get(1)).contains(infraStr).contains(MY_SINK);
1102
1103         // repeat with a response
1104         appender.clearExtractions();
1105         oper.logMessage(EventType.IN, SOURCE_INFRA, MY_SOURCE, new MyData());
1106         output = appender.getExtracted();
1107         assertEquals(2, output.size());
1108         assertThat(output.get(0)).contains("cannot pretty-print response");
1109         assertThat(output.get(1)).contains(MY_SOURCE);
1110     }
1111
1112     @Test
1113     public void testGetRetry() {
1114         assertEquals(0, oper.getRetry(null));
1115         assertEquals(10, oper.getRetry(10));
1116     }
1117
1118     @Test
1119     public void testGetRetryWait() {
1120         // need an operator that doesn't override the retry time
1121         OperationPartial oper2 = new OperationPartial(params, config) {};
1122         assertEquals(OperationPartial.DEFAULT_RETRY_WAIT_MS, oper2.getRetryWaitMs());
1123     }
1124
1125     @Test
1126     public void testGetTimeOutMs() {
1127         assertEquals(TIMEOUT * 1000, oper.getTimeoutMs(params.getTimeoutSec()));
1128
1129         params = params.toBuilder().timeoutSec(null).build();
1130
1131         // new params, thus need a new operation
1132         oper = new MyOper();
1133
1134         assertEquals(0, oper.getTimeoutMs(params.getTimeoutSec()));
1135     }
1136
1137     private void starter(OperationOutcome oper) {
1138         ++numStart;
1139         tstart = oper.getStart();
1140         opstart = oper;
1141         starts.add(oper);
1142     }
1143
1144     private void completer(OperationOutcome oper) {
1145         ++numEnd;
1146         opend = oper;
1147         ends.add(oper);
1148     }
1149
1150     /**
1151      * Gets a function that does nothing.
1152      *
1153      * @param <T> type of input parameter expected by the function
1154      * @return a function that does nothing
1155      */
1156     private <T> Consumer<T> noop() {
1157         return unused -> {
1158         };
1159     }
1160
1161     private OperationOutcome makeSuccess() {
1162         OperationOutcome outcome = params.makeOutcome();
1163         outcome.setResult(PolicyResult.SUCCESS);
1164
1165         return outcome;
1166     }
1167
1168     private OperationOutcome makeFailure() {
1169         OperationOutcome outcome = params.makeOutcome();
1170         outcome.setResult(PolicyResult.FAILURE);
1171
1172         return outcome;
1173     }
1174
1175     /**
1176      * Verifies a run.
1177      *
1178      * @param testName test name
1179      * @param expectedCallbacks number of callbacks expected
1180      * @param expectedOperations number of operation invocations expected
1181      * @param expectedResult expected outcome
1182      */
1183     private void verifyRun(String testName, int expectedCallbacks, int expectedOperations,
1184                     PolicyResult expectedResult) {
1185
1186         verifyRun(testName, expectedCallbacks, expectedOperations, expectedResult, noop());
1187     }
1188
1189     /**
1190      * Verifies a run.
1191      *
1192      * @param testName test name
1193      * @param expectedCallbacks number of callbacks expected
1194      * @param expectedOperations number of operation invocations expected
1195      * @param expectedResult expected outcome
1196      * @param manipulator function to modify the future returned by
1197      *        {@link OperationPartial#start(ControlLoopOperationParams)} before the tasks
1198      *        in the executor are run
1199      */
1200     private void verifyRun(String testName, int expectedCallbacks, int expectedOperations, PolicyResult expectedResult,
1201                     Consumer<CompletableFuture<OperationOutcome>> manipulator) {
1202
1203         tstart = null;
1204         opstart = null;
1205         opend = null;
1206         starts.clear();
1207         ends.clear();
1208
1209         CompletableFuture<OperationOutcome> future = oper.start();
1210
1211         manipulator.accept(future);
1212
1213         assertTrue(testName, executor.runAll(MAX_REQUESTS));
1214
1215         assertEquals(testName, expectedCallbacks, numStart);
1216         assertEquals(testName, expectedCallbacks, numEnd);
1217
1218         if (expectedCallbacks > 0) {
1219             assertNotNull(testName, opstart);
1220             assertNotNull(testName, opend);
1221             assertEquals(testName, expectedResult, opend.getResult());
1222
1223             assertSame(testName, tstart, opstart.getStart());
1224             assertSame(testName, tstart, opend.getStart());
1225
1226             try {
1227                 assertTrue(future.isDone());
1228                 assertEquals(testName, opend, future.get());
1229
1230                 // "start" is never final
1231                 for (OperationOutcome outcome : starts) {
1232                     assertFalse(testName, outcome.isFinalOutcome());
1233                 }
1234
1235                 // only the last "complete" is final
1236                 assertTrue(testName, ends.removeLast().isFinalOutcome());
1237
1238                 for (OperationOutcome outcome : ends) {
1239                     assertFalse(outcome.isFinalOutcome());
1240                 }
1241
1242             } catch (InterruptedException | ExecutionException e) {
1243                 throw new IllegalStateException(e);
1244             }
1245
1246             if (expectedOperations > 0) {
1247                 assertNotNull(testName, oper.getSubRequestId());
1248                 assertEquals(testName + " op start", oper.getSubRequestId(), opstart.getSubRequestId());
1249                 assertEquals(testName + " op end", oper.getSubRequestId(), opend.getSubRequestId());
1250             }
1251         }
1252
1253         assertEquals(testName, expectedOperations, oper.getCount());
1254     }
1255
1256     /**
1257      * Creates a new {@link #oper} whose coder will throw an exception.
1258      */
1259     private void setOperCoderException() {
1260         oper = new MyOper() {
1261             @Override
1262             protected Coder getCoder() {
1263                 return new StandardCoder() {
1264                     @Override
1265                     public String encode(Object object, boolean pretty) throws CoderException {
1266                         throw new CoderException(EXPECTED_EXCEPTION);
1267                     }
1268                 };
1269             }
1270         };
1271     }
1272
1273
1274     @Getter
1275     public static class MyData {
1276         private String text = TEXT;
1277     }
1278
1279
1280     private class MyOper extends OperationPartial {
1281         @Getter
1282         private int count = 0;
1283
1284         @Setter
1285         private boolean genException;
1286         @Setter
1287         private int maxFailures = 0;
1288         @Setter
1289         private CompletableFuture<OperationOutcome> preProc;
1290
1291
1292         public MyOper() {
1293             super(OperationPartialTest.this.params, config);
1294         }
1295
1296         @Override
1297         protected OperationOutcome doOperation(int attempt, OperationOutcome operation) {
1298             ++count;
1299             if (genException) {
1300                 throw new IllegalStateException(EXPECTED_EXCEPTION);
1301             }
1302
1303             operation.setSubRequestId(String.valueOf(attempt));
1304
1305             if (count > maxFailures) {
1306                 operation.setResult(PolicyResult.SUCCESS);
1307             } else {
1308                 operation.setResult(PolicyResult.FAILURE);
1309             }
1310
1311             return operation;
1312         }
1313
1314         @Override
1315         protected long getRetryWaitMs() {
1316             /*
1317              * Sleep timers run in the background, but we want to control things via the
1318              * "executor", thus we avoid sleep timers altogether by simply returning 0.
1319              */
1320             return 0L;
1321         }
1322
1323         @Override
1324         protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
1325             return (preProc != null ? preProc : super.startPreprocessorAsync());
1326         }
1327     }
1328 }