### Question:
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() )); } } }  Pair<AbstractStreamCreateStatement, String> checkAndSetAvroSchema(
      final AbstractStreamCreateStatement abstractStreamCreateStatement,
      final Map<String, Object> streamsProperties,
      final SchemaRegistryClient schemaRegistryClient
  ); void validatePersistentQueryResults(
      final PersistentQueryMetadata persistentQueryMetadata,
      final SchemaRegistryClient schemaRegistryClient
  );  }### Answer:
@Test public void shouldValidatePersistentQueryResultCorrectly() throws IOException, RestClientException { SchemaRegistryClient schemaRegistryClient = mock(SchemaRegistryClient.class); KsqlTopic resultTopic = new KsqlTopic("testTopic", "testTopic", new KsqlAvroTopicSerDe()); Schema resultSchema = SerDeUtil.getSchemaFromAvro(ordersAveroSchemaStr); PersistentQueryMetadata persistentQueryMetadata = buildStubPersistentQueryMetadata(resultSchema, resultTopic); org.apache.avro.Schema.Parser parser = new org.apache.avro.Schema.Parser(); org.apache.avro.Schema avroSchema = parser.parse(ordersAveroSchemaStr); expect(schemaRegistryClient.testCompatibility(anyString(), EasyMock.isA(avroSchema.getClass()))) .andReturn(true); replay(schemaRegistryClient); avroUtil.validatePersistentQueryResults(persistentQueryMetadata, schemaRegistryClient); }
@Test public void shouldFailForInvalidResultAvroSchema() throws IOException, RestClientException { SchemaRegistryClient schemaRegistryClient = mock(SchemaRegistryClient.class); KsqlTopic resultTopic = new KsqlTopic("testTopic", "testTopic", new KsqlAvroTopicSerDe ()); Schema resultSchema = SerDeUtil.getSchemaFromAvro(ordersAveroSchemaStr); PersistentQueryMetadata persistentQueryMetadata = buildStubPersistentQueryMetadata(resultSchema, resultTopic); expect(schemaRegistryClient.testCompatibility(anyString(), anyObject())).andReturn(false); replay(schemaRegistryClient); try { avroUtil.validatePersistentQueryResults(persistentQueryMetadata, schemaRegistryClient); fail(); } catch (Exception e) { assertThat("Incorrect exception message", "Cannot register avro schema for testTopic since " + "it is not valid for schema registry.", equalTo(e.getMessage())); } }