static Properties resolveConsumerProperties(Map<String, String> options, Object keySchema, Object valueSchema) {
        Properties properties = from(options);

        withSerdeConsumerProperties(true, options, keySchema, properties);
        withSerdeConsumerProperties(false, options, valueSchema, properties);

        return properties;
    }
@Test
    public void test_consumerProperties_avro_schemaRegistry() {
        // key
        assertThat(resolveConsumerProperties(Map.of(
                OPTION_KEY_FORMAT, AVRO_FORMAT,
                "schema.registry.url", "http://localhost:8081"
        ))).containsExactlyInAnyOrderEntriesOf(Map.of(
                KEY_DESERIALIZER, KafkaAvroDeserializer.class.getCanonicalName(),
                "schema.registry.url", "http://localhost:8081"
        ));

        // value
        assertThat(resolveConsumerProperties(Map.of(
                OPTION_KEY_FORMAT, UNKNOWN_FORMAT,
                OPTION_VALUE_FORMAT, AVRO_FORMAT,
                "schema.registry.url", "http://localhost:8081"
        ))).containsExactlyInAnyOrderEntriesOf(Map.of(
                VALUE_DESERIALIZER, KafkaAvroDeserializer.class.getCanonicalName(),
                "schema.registry.url", "http://localhost:8081"
        ));
    }