@Override
  public void waitFor(
      final KsqlEntityList previousCommands,
      final Class<? extends Statement> statementClass) {
    if (mustSync.test(statementClass)) {
      final ArrayList<KsqlEntity> reversed = new ArrayList<>(previousCommands);
      Collections.reverse(reversed);

      reversed.stream()
          .filter(e -> e instanceof CommandStatusEntity)
          .map(CommandStatusEntity.class::cast)
          .map(CommandStatusEntity::getCommandSequenceNumber)
          .findFirst()
          .ifPresent(seqNum -> {
            try {
              commandQueue.ensureConsumedPast(seqNum, timeout);
            } catch (final InterruptedException e) {
              throw new KsqlRestException(Errors.serverShuttingDown());
            } catch (final TimeoutException e) {
              throw new KsqlRestException(Errors.commandQueueCatchUpTimeout(seqNum));
            }
          });
    }
  }
@Test
  public void shouldWaitForDistributedStatements() throws Exception {
    // Given:
    givenSyncWithPredicate(clazz -> true);
    givenEntities(commandStatusEntity1);

    // When:
    commandQueueSync.waitFor(entities, CreateStreamAsSelect.class);

    // Then:
    verify(commandQueue, times(1)).ensureConsumedPast(1L, Duration.ZERO);
  }