@Override
  public Serializer serializer(String topic, Target type) {
    String subject = schemaSubject(topic, type);
    SchemaMetadata meta = getSchemaBySubject(subject)
        .orElseThrow(() -> new ValidationException(
            String.format("No schema for subject '%s' found", subject)));
    ParsedSchema schema = getSchemaById(meta.getId())
        .orElseThrow(() -> new IllegalStateException(
            String.format("Schema found for id %s, subject '%s'", meta.getId(), subject)));
    SchemaType schemaType = SchemaType.fromString(meta.getSchemaType())
        .orElseThrow(() -> new IllegalStateException("Unknown schema type: " + meta.getSchemaType()));
    return switch (schemaType) {
      case PROTOBUF -> input ->
          serializeProto(schemaRegistryClient, topic, type, (ProtobufSchema) schema, meta.getId(), input);
      case AVRO -> input ->
          serializeAvro((AvroSchema) schema, meta.getId(), input);
      case JSON -> input ->
          serializeJson((JsonSchema) schema, meta.getId(), input);
    };
  }
@Test
  void serializeTreatsInputAsJsonAvroSchemaPayload() throws RestClientException, IOException {
    AvroSchema schema = new AvroSchema(
        "{"
            + "  \"type\": \"record\","
            + "  \"name\": \"TestAvroRecord1\","
            + "  \"fields\": ["
            + "    {"
            + "      \"name\": \"field1\","
            + "      \"type\": \"string\""
            + "    },"
            + "    {"
            + "      \"name\": \"field2\","
            + "      \"type\": \"int\""
            + "    }"
            + "  ]"
            + "}"
    );
    String jsonValue = "{ \"field1\":\"testStr\", \"field2\": 123 }";
    String topic = "test";

    int schemaId = registryClient.register(topic + "-value", schema);
    byte[] serialized = serde.serializer(topic, Serde.Target.VALUE).serialize(jsonValue);
    byte[] expected = toBytesWithMagicByteAndSchemaId(schemaId, jsonValue, schema);
    assertThat(serialized).isEqualTo(expected);
  }