@Override
  public ParsedSchema fromConnectSchema(final Schema schema) {
    final Schema avroCompatibleSchema = AvroSchemas
        .getAvroCompatibleConnectSchema(schema, formatProps.getFullSchemaName());

    try {
      return new AvroSchema(avroData.fromConnectSchema(avroCompatibleSchema));
    } catch (final SchemaParseException e) {
      throw new KsqlException("Schema is not compatible with Avro: " + e.getMessage(), e);
    }
  }
@Test
  public void shouldThrowWhenBuildingAvroSchemafSchemaContainsInvalidAvroNames() {
    // Given:
    final Schema schema = SchemaBuilder.struct()
        .field("2Bad", Schema.OPTIONAL_INT32_SCHEMA)
        .build();

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> translator.fromConnectSchema(schema)
    );

    // Then:
    assertThat(e.getMessage(), is("Schema is not compatible with Avro: Illegal initial character: 2Bad"));
  }