@Override
  public ExecuteResult execute(final ServiceContext serviceContext, final ConfiguredKsqlPlan plan,
                               final boolean restoreInProgress) {
    try {
      final ExecuteResult result = EngineExecutor
          .create(primaryContext, serviceContext, plan.getConfig())
          .execute(plan.getPlan(), restoreInProgress);
      return result;
    } catch (final KsqlStatementException e) {
      throw e;
    } catch (final KsqlException e) {
      // add the statement text to the KsqlException
      throw new KsqlStatementException(
          e.getMessage(),
          e.getMessage(),
          plan.getPlan().getStatementText(),
          e.getCause()
      );
    }
  }
@Test
  public void shouldCreateSourceTablesQueries() {
    // Given:
    setupKsqlEngineWithSharedRuntimeEnabled();
    givenTopicsExist("t1_topic");

    // When:
    final List<QueryMetadata> queries = KsqlEngineTestUtil.execute(
        serviceContext,
        ksqlEngine,
        "create source table t1 (f0 bigint primary key, f1 double, f2 boolean) "
            + "with (kafka_topic='t1_topic', value_format='json');",
        ksqlConfig,
        Collections.emptyMap());

    // Then:
    assertThat(queries, hasSize(1));
    assertThat(queries.get(0), is(instanceOf(PersistentQueryMetadata.class)));

    final PersistentQueryMetadata metadata = (PersistentQueryMetadata) queries.get(0);
    assertThat(metadata.getPersistentQueryType(),
        is(KsqlConstants.PersistentQueryType.CREATE_SOURCE));
    assertThat((metadata).getSink(), is(Optional.empty()));
    assertThat((metadata).getSinkName(), is(Optional.empty()));
    assertThat((metadata).getDataSourceType(), is(Optional.empty()));
    assertThat((metadata).getResultTopic(), is(Optional.empty()));
    assertThat(metadata.getLogicalSchema().key(), hasItems(hasFullName("F0")));
    assertThat(metadata.getLogicalSchema().value(), hasItems(hasFullName("F1"), hasFullName("F2")));
  }