### Question:
SchemaUtil { public static String buildAvroSchema(final Schema schema, String name) { org.apache.avro.SchemaBuilder.FieldAssembler fieldAssembler = org.apache.avro.SchemaBuilder .record(name).namespace("ksql") .fields(); for (Field field : schema.fields()) { fieldAssembler .name(field.name().replace(".", "_")) .type(getAvroSchemaForField(field.schema())) .withDefault(null); } return fieldAssembler.endRecord().toString(); }  static Class getJavaType(final Schema schema); static Optional<Field> getFieldByName(final Schema schema, final String fieldName); static Schema getTypeSchema(final String sqlType); static int getFieldIndexByName(final Schema schema, final String fieldName); static Schema buildSchemaWithAlias(final Schema schema, final String alias); static String getSchemaFieldName(Field field); static String getJavaCastString(Schema schema); static Schema addImplicitRowTimeRowKeyToSchema(Schema schema); static Schema removeImplicitRowTimeRowKeyFromSchema(Schema schema); static Set<Integer> getRowTimeRowKeyIndexes(Schema schema); static String getSchemaDefinitionString(Schema schema); static String getSqlTypeName(Schema schema); static String buildAvroSchema(final Schema schema, String name); static Schema getAvroSerdeKsqlSchema(Schema schema); static String getFieldNameWithNoAlias(Field field); static Schema getSchemaWithNoAlias(Schema schema); static int getIndexInSchema(final String fieldName, final Schema schema); static final String ARRAY; static final String MAP; static final String ROWKEY_NAME; static final String ROWTIME_NAME; static final int ROWKEY_NAME_INDEX; }### Answer:
@Test public void shouldCreateCorrectAvroSchemaWithNullableFields() { SchemaBuilder schemaBuilder = SchemaBuilder.struct(); schemaBuilder .field("ordertime", Schema.INT64_SCHEMA) .field("orderid", Schema.STRING_SCHEMA) .field("itemid", Schema.STRING_SCHEMA) .field("orderunits", Schema.FLOAT64_SCHEMA) .field("arraycol", SchemaBuilder.array(Schema.FLOAT64_SCHEMA)) .field("mapcol", SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.FLOAT64_SCHEMA)); String avroSchemaString = SchemaUtil.buildAvroSchema(schemaBuilder.build(), "orders"); assertThat(avroSchemaString, equalTo( "{\"type\":\"record\",\"name\":\"orders\",\"namespace\":\"ksql\",\"fields\":" + "[{\"name\":\"ordertime\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":" + "\"orderid\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"itemid\"," + "\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"orderunits\",\"type\":" + "[\"null\",\"double\"],\"default\":null},{\"name\":\"arraycol\",\"type\":[\"null\"," + "{\"type\":\"array\",\"items\":[\"null\",\"double\"]}],\"default\":null},{\"name\":" + "\"mapcol\",\"type\":[\"null\",{\"type\":\"map\",\"values\":[\"null\",\"double\"]}]" + ",\"default\":null}]}")); }