@Override
  public Optional<SchemaDescription> getSchema(String topic, Target type) {
    String subject = schemaSubject(topic, type);
    return getSchemaBySubject(subject)
        .flatMap(schemaMetadata ->
            //schema can be not-found, when schema contexts configured improperly
            getSchemaById(schemaMetadata.getId())
                .map(parsedSchema ->
                    new SchemaDescription(
                        convertSchema(schemaMetadata, parsedSchema),
                        Map.of(
                            "subject", subject,
                            "schemaId", schemaMetadata.getId(),
                            "latestVersion", schemaMetadata.getVersion(),
                            "type", schemaMetadata.getSchemaType() // AVRO / PROTOBUF / JSON
                        )
                    )));
  }
@Test
  void returnsEmptyDescriptorIfSchemaNotRegisteredInSR() {
    String topic = "test";
    assertThat(serde.getSchema(topic, Serde.Target.KEY)).isEmpty();
    assertThat(serde.getSchema(topic, Serde.Target.VALUE)).isEmpty();
  }