### Question:
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 (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); } }  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 shouldPassAvroCheck() throws Exception { SchemaRegistryClient schemaRegistryClient = mock(SchemaRegistryClient.class); SchemaMetadata schemaMetadata = new SchemaMetadata(1, 1, ordersAveroSchemaStr); expect(schemaRegistryClient.getLatestSchemaMetadata(anyString())).andReturn(schemaMetadata); replay(schemaRegistryClient); AbstractStreamCreateStatement abstractStreamCreateStatement = getAbstractStreamCreateStatement ("CREATE STREAM S1 WITH " + "(kafka_topic='s1_topic', " + "value_format='avro' );"); Pair<AbstractStreamCreateStatement, String> checkResult = avroUtil.checkAndSetAvroSchema(abstractStreamCreateStatement, new HashMap<>(), schemaRegistryClient); AbstractStreamCreateStatement newAbstractStreamCreateStatement = checkResult.getLeft(); assertThat(newAbstractStreamCreateStatement.getElements(), equalTo(Arrays.asList( new TableElement("ORDERTIME", "BIGINT"), new TableElement("ORDERID", "BIGINT"), new TableElement("ITEMID", "VARCHAR"), new TableElement("ORDERUNITS", "DOUBLE"), new TableElement("ARRAYCOL", "ARRAY<DOUBLE>"), new TableElement("MAPCOL", "MAP<VARCHAR,DOUBLE>") ))); }
@Test public void shouldNotPassAvroCheckIfSchemaDoesNotExist() throws Exception { SchemaRegistryClient schemaRegistryClient = mock(SchemaRegistryClient.class); SchemaMetadata schemaMetadata = new SchemaMetadata(1, 1, null); expect(schemaRegistryClient.getLatestSchemaMetadata(anyString())).andReturn(schemaMetadata); replay(schemaRegistryClient); AbstractStreamCreateStatement abstractStreamCreateStatement = getAbstractStreamCreateStatement ("CREATE STREAM S1 WITH " + "(kafka_topic='s1_topic', " + "value_format='avro' );"); try { avroUtil.checkAndSetAvroSchema(abstractStreamCreateStatement, new HashMap<>(), schemaRegistryClient); fail(); } catch (Exception e) { assertThat("Expected different message message.", e.getMessage().trim(), equalTo("Unable to verify the AVRO schema is compatible with KSQL. null")); } }