public static void throwOnDisabledFeatures(final ConfiguredStatement<?> statement) {
    final SessionConfig sessionConfig = statement.getSessionConfig();

    if (statement.getStatement() instanceof CreateSource) {
      throwOnDisabledFeatures((CreateSource) statement.getStatement(), sessionConfig);
    }
  }
@Test
  public void shouldNotThrowOnCreateStreamWithHeadersIfFeatureFlagIsEnabled() {
    // Given
    final KsqlConfig config = new KsqlConfig(ImmutableMap.of(
        KsqlConfig.KSQL_HEADERS_COLUMNS_ENABLED, true
    ));

    final CreateStream createStream = new CreateStream(
        SourceName.of("stream1"),
        TableElements.of(new TableElement(
            ColumnName.of("col1"),
            new Type(SqlTypes.INTEGER),
            new ColumnConstraints.Builder().headers().build()
        )),
        false,
        false,
        CreateSourceProperties.from(ImmutableMap.of(
            CommonCreateConfigs.KAFKA_TOPIC_NAME_PROPERTY, new StringLiteral("topic1")
        )),
        false
    );
    final ConfiguredStatement<?> configuredStatement = configured(config, createStream);

    // When/Then
    FeatureFlagChecker.throwOnDisabledFeatures(configuredStatement);
  }