@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 shouldRunPushQuery_success() {
    // Given:
    when(ksqlEngine.executeTransientQuery(any(), any(), anyBoolean()))
        .thenReturn(transientQueryMetadata);

    // When:
    final QueryMetadataHolder queryMetadataHolder = queryExecutor.handleStatement(
        serviceContext, ImmutableMap.of(), ImmutableMap.of(),
        pushQuery, Optional.empty(), metricsCallbackHolder, context, false);
    // Should be no metrics reported for push queries
    metricsCallbackHolder.reportMetrics(200, 1000L, 5000L, 20000L);

    // Then:
    verifyNoMoreInteractions(pullQueryExecutorMetrics);
    assertThat(queryMetadataHolder.getTransientQueryMetadata().isPresent(), is(true));
    verify(localCommands).write(transientQueryMetadata);
  }