class AvroUtil {

  public void validatePersistentQueryResults(
      final PersistentQueryMetadata persistentQueryMetadata,
      final SchemaRegistryClient schemaRegistryClient
  ) {

    if (persistentQueryMetadata.getResultTopicSerde() == DataSource.DataSourceSerDe.AVRO) {
      String avroSchemaString = SchemaUtil.buildAvroSchema(
          persistentQueryMetadata.getResultSchema(),
          persistentQueryMetadata.getResultTopic().getName()
      );
      boolean isValidSchema = isValidAvroSchemaForTopic(
          persistentQueryMetadata.getResultTopic().getTopicName(),
          avroSchemaString,
          schemaRegistryClient
      );
      if (!isValidSchema) {
        throw new KsqlException(String.format(
            "Cannot register avro schema for %s since it is not valid for schema registry.",
            persistentQueryMetadata.getResultTopic().getKafkaTopicName()
        ));
      }
    }
  }

}