@Override
  public void close() {
  }
@Test
  public void shouldSucceed_gapDetectedRemote_retry()
      throws ExecutionException, InterruptedException {
    // Given:
    final AtomicReference<Set<KsqlNode>> nodes = new AtomicReference<>(
        ImmutableSet.of(ksqlNodeLocal, ksqlNodeRemote));
    final PushRouting routing = new PushRouting(sqr -> nodes.get(), 50, true);
    AtomicReference<TestRemotePublisher> remotePublisher = new AtomicReference<>();
    AtomicInteger remoteCount = new AtomicInteger(0);
    when(simpleKsqlClient.makeQueryRequestStreamed(any(), any(), any(), any()))
        .thenAnswer(a -> {
          remotePublisher.set(new TestRemotePublisher(context));
          remoteCount.incrementAndGet();
          final Map<String, ?> requestProperties = a.getArgument(3);
          String continuationToken = (String) requestProperties.get(
              KsqlRequestConfig.KSQL_REQUEST_QUERY_PUSH_CONTINUATION_TOKEN);
          if (remoteCount.get() == 1) {
            assertThat(continuationToken, nullValue());
          } else if (remoteCount.get()  == 2) {
            assertThat(continuationToken, notNullValue());
            final PushOffsetRange range = PushOffsetRange.deserialize(continuationToken);
            assertThat(range.getEndOffsets().getDenseRepresentation(),
                is(ImmutableList.of(0L, 3L)));
            remotePublisher.get().accept(REMOTE_ROW2);
          }
          return createFuture(RestResponse.successful(200, remotePublisher.get()));
        });

    // When:
    final PushConnectionsHandle handle = handlePushRouting(routing);
    final AtomicReference<Throwable> exception = new AtomicReference<>(null);
    handle.onException(exception::set);
    context.runOnContext(v -> {
      remotePublisher.get().accept(REMOTE_CONTINUATION_TOKEN1);
      remotePublisher.get().accept(REMOTE_ROW1);
      remotePublisher.get().accept(REMOTE_CONTINUATION_TOKEN_GAP);
    });

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

    // Then:
    verify(simpleKsqlClient, times(2)).makeQueryRequestStreamed(any(), any(), any(), any());
    assertThat(rows.contains(REMOTE_ROW1.getRow().get().getColumns()), is(true));
    assertThat(rows.contains(REMOTE_ROW2.getRow().get().getColumns()), is(true));
  }