### Question:
SerDeUtil { public static Schema getSchemaFromAvro(String avroSchemaString) { org.apache.avro.Schema.Parser parser = new org.apache.avro.Schema.Parser(); org.apache.avro.Schema avroSchema = parser.parse(avroSchemaString); SchemaBuilder inferredSchema = SchemaBuilder.struct().name(avroSchema.getName()); for (org.apache.avro.Schema.Field avroField: avroSchema.getFields()) { inferredSchema.field(avroField.name(), getKsqlSchemaForAvroSchema(avroField.schema())); } return inferredSchema.build(); }  static Schema getSchemaFromAvro(String avroSchemaString);  }### Answer:
@Test public void shouldReturnCorrectKsqlSchema() { Schema schema = SerDeUtil.getSchemaFromAvro(avroSchemaStr); assertThat("Incorrect schema.", schema.fields().size(), equalTo(8)); assertThat("Incorrect field schema.", schema.fields().get(0).schema(), equalTo(Schema.STRING_SCHEMA)); assertThat("Incorrect field schema.", schema.fields().get(1).schema(), equalTo(Schema .STRING_SCHEMA)); assertThat("Incorrect field schema.", schema.fields().get(2).schema(), equalTo(Schema .STRING_SCHEMA)); assertThat("Incorrect field schema.", schema.fields().get(3).schema(), equalTo(Schema .STRING_SCHEMA)); assertThat("Incorrect field schema.", schema.fields().get(4).schema(), equalTo(Schema .STRING_SCHEMA)); assertThat("Incorrect field schema.", schema.fields().get(5).schema(), equalTo(Schema .FLOAT64_SCHEMA)); assertThat("Incorrect field schema.", schema.fields().get(6).schema(), equalTo(Schema .FLOAT64_SCHEMA)); assertThat("Incorrect field schema.", schema.fields().get(7).schema(), equalTo(Schema .STRING_SCHEMA)); }
@Test public void shouldFailForUnsupportedUnion() { String avroSchemaStr = "{\n" + " \"type\": \"record\",\n" + " \"name\": \"LOGON\",\n" + " \"namespace\": \"ORCL.SOE2\",\n" + " \"fields\": [\n" + " {\n" + " \"name\": \"table\",\n" + " \"type\": [\n" + " \"string\"\n" + " ],\n" + " \"default\": null\n" + " },\n" + " {\n" + " \"name\": \"LOGON_DATE\",\n" + " \"type\": [\n" + " \"null\",\n" + " \"int\",\n" + " \"long\"\n" + " ],\n" + " \"default\": null\n" + " }\n" + " ],\n" + " \"connect.name\": \"ORCL.SOE2.LOGON\"\n" + "}"; try { Schema schema = SerDeUtil.getSchemaFromAvro(avroSchemaStr); fail(); } catch (KsqlException ksqlException) { assertThat(ksqlException.getMessage(), equalTo("Union type cannot have more than two " + "types and one of them should be null.")); } }
@Test public void shouldFailForUnionWithNoNull() { String avroSchemaStr = "{\n" + " \"type\": \"record\",\n" + " \"name\": \"LOGON\",\n" + " \"namespace\": \"ORCL.SOE2\",\n" + " \"fields\": [\n" + " {\n" + " \"name\": \"table\",\n" + " \"type\": [\n" + " \"string\"\n" + " ],\n" + " \"default\": null\n" + " },\n" + " {\n" + " \"name\": \"LOGON_DATE\",\n" + " \"type\": [\n" + " \"string\",\n" + " \"long\"\n" + " ],\n" + " \"default\": null\n" + " }\n" + " ],\n" + " \"connect.name\": \"ORCL.SOE2.LOGON\"\n" + "}"; try { Schema schema = SerDeUtil.getSchemaFromAvro(avroSchemaStr); fail(); } catch (KsqlException ksqlException) { assertThat(ksqlException.getMessage(), equalTo("Union type cannot have more than two " + "types and one of them should be null.")); } }