@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 shouldThrowOnInsertIntoWithKeyMismatch() {
    // Given:
    setupKsqlEngineWithSharedRuntimeEnabled();

    execute(
        serviceContext,
        ksqlEngine,
        "create stream bar as select * from orders;",
        ksqlConfig,
        emptyMap()
    );

    // When:
    final KsqlStatementException e = assertThrows(
        KsqlStatementException.class,
        () -> KsqlEngineTestUtil.execute(
            serviceContext,
            ksqlEngine,
            "insert into bar select * from orders partition by orderid;",
            ksqlConfig,
            Collections.emptyMap()
        )
    );

    // Then:
    assertThat(e, rawMessage(containsString("Incompatible schema between results and sink.")));
    assertThat(e, rawMessage(containsString("Result schema is `ORDERID` BIGINT KEY, ")));
    assertThat(e, rawMessage(containsString("Sink schema is `ORDERTIME` BIGINT KEY, ")));

    assertThat(e, statementText(is(
        "insert into bar select * from orders partition by orderid;")));
  }