@Override
  public void close() {
  }
@Test
  public void shouldSucceed_gapDetectedRemote_noRetry()
      throws ExecutionException, InterruptedException {
    // Given:
    final AtomicReference<Set<KsqlNode>> nodes = new AtomicReference<>(
        ImmutableSet.of(ksqlNodeLocal, ksqlNodeRemote));
    final PushRouting routing = new PushRouting(sqr -> nodes.get(), 50, false);

    // When:
    final PushConnectionsHandle handle = handlePushRouting(routing);
    context.runOnContext(v -> {
      remotePublisher.accept(REMOTE_ROW1);
      remotePublisher.accept(REMOTE_CONTINUATION_TOKEN1);
      remotePublisher.accept(REMOTE_ROW2);
      remotePublisher.accept(REMOTE_CONTINUATION_TOKEN_GAP);
    });

    Set<List<?>> rows = waitOnRows(2);
    waitOnNodeStatus(handle, ksqlNodeRemote, RoutingResultStatus.OFFSET_GAP_FOUND);
    handle.close();

    // Then:
    assertThat(rows.contains(REMOTE_ROW1.getRow().get().getColumns()), is(true));
    assertThat(rows.contains(REMOTE_ROW2.getRow().get().getColumns()), is(true));
    assertThat(handle.get(ksqlNodeRemote).get().getStatus(),
        is(RoutingResultStatus.OFFSET_GAP_FOUND));
  }