class AvroUtil {

  public static void throwOnInvalidSchemaEvolution(
      final String statementText,
      final CreateSourceCommand ddl,
      final SchemaRegistryClient schemaRegistryClient
  ) {
    final Formats formats = ddl.getFormats();
    final FormatInfo format = formats.getValueFormat();
    if (FormatFactory.of(format) != FormatFactory.AVRO) {
      return;
    }

    final SchemaTranslator translator = new AvroFormat()
        .getSchemaTranslator(format.getProperties());

    final AvroSchema parsedSchema = (AvroSchema) translator.toParsedSchema(
        PersistenceSchema.from(
            ddl.getSchema().value(),
            formats.getValueFeatures()
        )
    );

    final String topicName = ddl.getTopicName();

    if (!isValidAvroSchemaForTopic(topicName, parsedSchema, schemaRegistryClient)) {
      throw new KsqlStatementException(String.format(
          "Cannot register avro schema for %s as the schema is incompatible with the current "
              + "schema version registered for the topic.%n"
              + "KSQL schema: %s%n"
              + "Registered schema: %s",
          topicName,
          parsedSchema,
          getRegisteredSchema(topicName, schemaRegistryClient)
      ), statementText);
    }
  }

  private  AvroUtil();

}