public static void waitForCommandSequenceNumber(
      final CommandQueue commandQueue,
      final KsqlRequest request,
      final Duration timeout
  ) throws InterruptedException, TimeoutException {
    final Optional<Long> commandSequenceNumber = request.getCommandSequenceNumber();
    if (commandSequenceNumber.isPresent()) {
      final long seqNum = commandSequenceNumber.get();
      commandQueue.ensureConsumedPast(seqNum, timeout);
    }
  }
@Test
  public void shouldWaitIfSequenceNumberSpecified() throws Exception {
    // Given:
    when(request.getCommandSequenceNumber()).thenReturn(Optional.of(SEQUENCE_NUMBER));

    // When:
    CommandStoreUtil.waitForCommandSequenceNumber(commandQueue, request, TIMEOUT);

    // Then:
    verify(commandQueue).ensureConsumedPast(SEQUENCE_NUMBER, TIMEOUT);
  }