@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 shouldInsertValuesIntoHeaderSchemaValueColumns() {
    // Given:
    givenSourceStreamWithSchema(SCHEMA_WITH_HEADERS, SerdeFeatures.of(), SerdeFeatures.of());
    final ConfiguredStatement<InsertValues> statement = givenInsertValues(
        ImmutableList.of(K0, COL0, COL1),
        ImmutableList.of(
            new StringLiteral("key"),
            new StringLiteral("str"),
            new LongLiteral(2L)
        )
    );

    // When:
    executor.execute(statement, mock(SessionProperties.class), engine, serviceContext);

    // Then:
    verify(producer).send(new ProducerRecord<>(TOPIC_NAME, null, 1L, KEY, VALUE));
  }