@SuppressWarnings("unchecked")
  public QueryMetadataHolder handleStatement(
      final ServiceContext serviceContext,
      final Map<String, Object> configOverrides,
      final Map<String, Object> requestProperties,
      final PreparedStatement<?> statement,
      final Optional<Boolean> isInternalRequest,
      final MetricsCallbackHolder metricsCallbackHolder,
      final Context context,
      final boolean excludeTombstones
  ) {

    if (statement.getStatement() instanceof Query) {
      return handleQuery(
          serviceContext,
          (PreparedStatement<Query>) statement,
          isInternalRequest,
          metricsCallbackHolder,
          configOverrides,
          requestProperties,
          context,
          excludeTombstones
      );
    } else {
      return QueryMetadataHolder.unhandled();
    }
  }
@Test
  public void shouldReachConcurrentLimitTablePullQuery() {
    // Given:
    when(concurrencyLimiter.increment()).thenThrow(new KsqlException("concurrencyLimiter Error!"));
    when(mockDataSource.getDataSourceType()).thenReturn(DataSourceType.KTABLE);

    // When:
    Exception e = assertThrows(KsqlException.class, () ->
        queryExecutor.handleStatement(serviceContext, ImmutableMap.of(), ImmutableMap.of(),
            pullQuery, Optional.empty(), metricsCallbackHolder, context, false));

    // Then:
    assertThat(e.getMessage(), containsString("concurrencyLimiter Error!"));
  }