class AvroUtil {

  public Pair<AbstractStreamCreateStatement, String> checkAndSetAvroSchema(
      final AbstractStreamCreateStatement abstractStreamCreateStatement,
      final Map<String, Object> streamsProperties,
      final SchemaRegistryClient schemaRegistryClient
  ) {

    Map<String, Expression> ddlProperties = abstractStreamCreateStatement.getProperties();
    if (!ddlProperties.containsKey(DdlConfig.VALUE_FORMAT_PROPERTY)) {
      throw new KsqlException(String.format(
          "%s should be set in WITH clause of CREATE STREAM/TABLE statement.",
          DdlConfig.VALUE_FORMAT_PROPERTY
      )
      );
    }
    final String serde = StringUtil.cleanQuotes(
        ddlProperties.get(DdlConfig.VALUE_FORMAT_PROPERTY).toString()
    );
    if (!serde.equalsIgnoreCase(DataSource.AVRO_SERDE_NAME)) {
      return new Pair<>(abstractStreamCreateStatement, null);
    }

    String kafkaTopicName = StringUtil.cleanQuotes(
        ddlProperties.get(DdlConfig.KAFKA_TOPIC_NAME_PROPERTY).toString()
    );
    try {
      // If the schema is not specified infer it from the Avro schema in Schema Registry.
      if (abstractStreamCreateStatement.getElements().isEmpty()) {
        SchemaMetadata schemaMetadata = fetchSchemaMetadata(
            abstractStreamCreateStatement,
            schemaRegistryClient,
            kafkaTopicName
        );

        String avroSchemaString = schemaMetadata.getSchema();
        streamsProperties.put(DdlConfig.AVRO_SCHEMA, avroSchemaString);
        Schema schema = SerDeUtil.getSchemaFromAvro(avroSchemaString);
        AbstractStreamCreateStatement abstractStreamCreateStatementCopy = addAvroFields(
            abstractStreamCreateStatement,
            schema,
            schemaMetadata.getId()
        );
        return new Pair<>(
            abstractStreamCreateStatementCopy,
            SqlFormatter.formatSql(abstractStreamCreateStatementCopy)
        );
      } else {
        return new Pair<>(abstractStreamCreateStatement, null);
      }
    } catch (Exception e) {
      String errorMessage = String.format(
          " Unable to verify the AVRO schema is compatible with KSQL. %s ",
          e.getMessage()
      );
      throw new KsqlException(errorMessage);
    }
  }

}