@SuppressWarnings("unused") // Part of required API.
  public void execute(
      final ConfiguredStatement<InsertValues> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    final InsertValues insertValues = statement.getStatement();
    final MetaStore metaStore = executionContext.getMetaStore();
    final KsqlConfig config = statement.getSessionConfig().getConfig(true);

    final DataSource dataSource = getDataSource(config, metaStore, insertValues);

    validateInsert(insertValues.getColumns(), dataSource);

    final ProducerRecord<byte[], byte[]> record =
        buildRecord(statement, metaStore, dataSource, serviceContext);

    try {
      producer.sendRecord(record, serviceContext, config.getProducerClientConfigProps());
    } catch (final TopicAuthorizationException e) {
      // TopicAuthorizationException does not give much detailed information about why it failed,
      // except which topics are denied. Here we just add the ACL to make the error message
      // consistent with other authorization error messages.
      final Exception rootCause = new KsqlTopicAuthorizationException(
          AclOperation.WRITE,
          e.unauthorizedTopics()
      );

      throw new KsqlException(createInsertFailedExceptionMessage(insertValues), rootCause);
    } catch (final ClusterAuthorizationException e) {
      // ClusterAuthorizationException is thrown when using idempotent producers
      // and either a topic write permission or a cluster-level idempotent write
      // permission (only applicable for broker versions no later than 2.8) is
      // missing. In this case, we include additional context to help the user
      // distinguish this type of failure from other permissions exceptions
      // such as the ones thrown above when TopicAuthorizationException is caught.
      throw new KsqlException(
          createInsertFailedExceptionMessage(insertValues),
          createClusterAuthorizationExceptionRootCause(dataSource)
      );
    } catch (final KafkaException e) {
      if (e.getCause() != null && e.getCause() instanceof ClusterAuthorizationException) {
        // The error message thrown when an idempotent producer is missing permissions
        // is (nondeterministically) inconsistent: it is either a raw ClusterAuthorizationException,
        // as checked for above, or a ClusterAuthorizationException wrapped inside a KafkaException.
        // ksqlDB handles these two the same way, accordingly.
        // See https://issues.apache.org/jira/browse/KAFKA-14138 for more.
        throw new KsqlException(
            createInsertFailedExceptionMessage(insertValues),
            createClusterAuthorizationExceptionRootCause(dataSource)
        );
      } else {
        throw new KsqlException(createInsertFailedExceptionMessage(insertValues), e);
      }
    } catch (final Exception e) {
      throw new KsqlException(createInsertFailedExceptionMessage(insertValues), e);
    }
  }
@Test
  public void shouldNotAllowInsertOnMultipleKeySchemaDefinitionsWithDifferentOrder() throws Exception {
    final String protoMultiSchema = "syntax = \"proto3\";\n"
        + "package io.proto;\n"
        + "\n"
        + "message SingleKey {\n"
        + "  string k0 = 1;\n"
        + "}\n"
        + "message MultiKeys {\n"
        + "  string k1 = 1;\n"
        + "  string k0 = 2;\n"
        + "}\n";

    // Given:
    when(srClient.getLatestSchemaMetadata(Mockito.any()))
        .thenReturn(new SchemaMetadata(1, 1, protoMultiSchema));
    when(srClient.getSchemaById(1))
        .thenReturn(new ProtobufSchema(protoMultiSchema));
    givenDataSourceWithSchema(
        TOPIC_NAME,
        SCHEMA_WITH_MUTI_KEYS,
        SerdeFeatures.of(SerdeFeature.SCHEMA_INFERENCE),
        SerdeFeatures.of(),
        FormatInfo.of(FormatFactory.PROTOBUF.name(), ImmutableMap.of(
            AvroProperties.FULL_SCHEMA_NAME,"io.proto.MultiKeys",
            AvroProperties.SCHEMA_ID, "1"
        )),
        FormatInfo.of(FormatFactory.JSON.name()),
        false,
        false);

    final ConfiguredStatement<InsertValues> statement = givenInsertValues(
        ImmutableList.of(K1, K0, COL0, COL1),
        ImmutableList.of(
            new StringLiteral("K1"),
            new StringLiteral("K0"),
            new StringLiteral("V0"),
            new LongLiteral(21))
    );

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> executor.execute(statement, mock(SessionProperties.class), engine, serviceContext)
    );

    // Then:
    assertThat(e.getMessage(), containsString("ksqlDB generated schema would overwrite existing key schema"));
    assertThat(e.getMessage(), containsString("Existing Schema: [`K1` STRING, `K0` STRING]"));
    assertThat(e.getMessage(), containsString("ksqlDB Generated: [`k0` STRING KEY, `k1` STRING KEY]"));
  }