@Override
  public void close() {
  }
@Test
  public void shouldSucceed_gapDetectedLocal_disableAlos()
      throws ExecutionException, InterruptedException {
    // Given:
    when(pushRoutingOptions.alosEnabled()).thenReturn(false);
    final AtomicReference<Set<KsqlNode>> nodes = new AtomicReference<>(
        ImmutableSet.of(ksqlNodeLocal, ksqlNodeRemote));
    final PushRouting routing = new PushRouting(sqr -> nodes.get(), 50, true);
    AtomicReference<TestLocalPublisher> localPublisher = new AtomicReference<>();
    AtomicInteger localCount = new AtomicInteger(0);
    when(pushPhysicalPlanManager.execute()).thenAnswer(a -> {
      localPublisher.set(new TestLocalPublisher(context));
      localCount.incrementAndGet();
      context.runOnContext(v -> {
        localPublisher.get().accept(LOCAL_ROW2);
      });
      return localPublisher.get();
    });
    doAnswer(a -> {
      final Optional<PushOffsetRange> newOffsetRange = a.getArgument(0);
      assertThat(newOffsetRange.isPresent(), is(true));
      assertThat(newOffsetRange.get().getEndOffsets(), is(ImmutableList.of(0L, 3L)));
      return null;
    }).when(pushPhysicalPlanManager).reset(any());

    // When:
    final PushConnectionsHandle handle = handlePushRouting(routing);
    context.runOnContext(v -> {
      localPublisher.get().accept(LOCAL_CONTINUATION_TOKEN1);
      localPublisher.get().accept(LOCAL_ROW1);
      localPublisher.get().accept(LOCAL_CONTINUATION_TOKEN_GAP);
    });

    Set<List<?>> rows = waitOnRows(2);
    handle.close();

    // Then:
    verify(pushPhysicalPlanManager, times(1)).execute();
    assertThat(rows.contains(LOCAL_ROW1.value().values()), is(true));
    assertThat(rows.contains(LOCAL_ROW2.value().values()), is(true));
  }