title: Kafka consumer unit test with Avro Schema registry failing
body: <p>I'm writing a consumer which listens to a Kafka topic and consumes message whenever message is available. I've tested the logic/code by running Kafka locally and it's working fine. </p>
  
  <p>While writing the unit/component test cases, it's failing with avro schema registry url error. I've tried different options available on internet but could not find anything working. I am not sure if my approach is even correct. Please help.</p>
  
  <p>Listener Class</p>
  
  <pre><code>@KafkaListener(topics = "positionmgmt.v1", containerFactory = "genericKafkaListenerFactory")
      public void receive(ConsumerRecord&lt;String, GenericRecord&gt; consumerRecord) {
          try {
              GenericRecord generic = consumerRecord.value();
              Object obj = generic.get("metadata");
  
              ObjectMapper mapper = new ObjectMapper();
  
              Header headerMetaData = mapper.readValue(obj.toString(), Header.class);
  
              System.out.println("Received payload :   " + consumerRecord.value());
  
              //Call backend with details in GenericRecord 
  
          }catch (Exception e){
              System.out.println("Exception while reading message from Kafka " + e );
          }
  </code></pre>
  
  <p>Kafka config </p>
  
  <pre><code>@Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, GenericRecord&gt; genericKafkaListenerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, GenericRecord&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(genericConsumerFactory());
          return factory;
      }
  
  public ConsumerFactory&lt;String, GenericRecord&gt; genericConsumerFactory() {
          Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
  
          config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
          config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");
          config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
          config.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG,"http://localhost:8081");
          return new DefaultKafkaConsumerFactory&lt;&gt;(config);
      }
  </code></pre>
  
  <p>Avro Schema</p>
  
  <pre><code>{
     "type":"record",
     "name":"KafkaEvent",
     "namespace":"com.ms.model.avro",
     "fields":[
        {
           "name":"metadata",
           "type":{
              "name":"metadata",
              "type":"record",
              "fields":[
                 {
                    "name":"correlationid",
                    "type":"string",
                    "doc":"this is corrleation id for transaction"
                 },
                 {
                    "name":"subject",
                    "type":"string",
                    "doc":"this is subject for transaction"
                 },
                 {
                    "name":"version",
                    "type":"string",
                    "doc":"this is version for transaction"
                 }
              ]
           }
        },
        {
           "name":"name",
           "type":"string"
        },
        {
           "name":"dept",
           "type":"string"
        },
        {
           "name":"empnumber",
           "type":"string"
        }
     ]
  }
  </code></pre>
  
  <p>This is my test code which I tried...</p>
  
  <pre><code>@ComponentTest
      @RunWith(SpringRunner.class)
      @EmbeddedKafka(partitions = 1, topics = { "positionmgmt.v1" })
      @SpringBootTest(classes={Application.class})
      @DirtiesContext
      public class ConsumeKafkaMessageTest {
  
        private static final String TEST_TOPIC = "positionmgmt.v1";
  
        @Autowired(required=true)
        EmbeddedKafkaBroker embeddedKafkaBroker;
  
        private Schema schema;
  
        private  SchemaRegistryClient schemaRegistry;
        private  KafkaAvroSerializer avroSerializer;
        private  KafkaAvroDeserializer avroDeserializer;
  
        private MockSchemaRegistryClient mockSchemaRegistryClient = new MockSchemaRegistryClient();
        private String registryUrl = "unused";
  
        private String avroSchema = string representation of avro schema
  
        @BeforeEach
        public void setUp() throws Exception {
          Schema.Parser parser = new Schema.Parser();
          schema = parser.parse(avroSchema);
  
          mockSchemaRegistryClient.register("Vendors-value", schema);
        }
  
        @Test
        public void consumeKafkaMessage_receive_sucess() {
  
          Schema metadataSchema = schema.getField("metadata").schema();
          GenericRecord metadata = new GenericData.Record(metadataSchema);
          metadata.put("version", "1.0");
          metadata.put("correlationid", "correlationid");
          metadata.put("subject", "metadata");
  
          GenericRecord record = new GenericData.Record(schema);
          record.put("metadata", metadata);
          record.put("name", "ABC");
          record.put("dept", "XYZ");
  
          Consumer&lt;String, GenericRecord&gt; consumer = configureConsumer();
          Producer&lt;String, GenericRecord&gt; producer = configureProducer();
  
          ProducerRecord&lt;String, GenericRecord&gt; prodRecord = new ProducerRecord&lt;String, GenericRecord&gt;(TEST_TOPIC, record);
  
          producer.send(prodRecord);
  
          ConsumerRecord&lt;String, GenericRecord&gt; singleRecord = KafkaTestUtils.getSingleRecord(consumer, TEST_TOPIC);
          assertNotNull(singleRecord.value());
  
          consumer.close();
          producer.close();
  
        }
  
        private Consumer&lt;String, GenericRecord&gt; configureConsumer() {
          Map&lt;String, Object&gt; consumerProps = KafkaTestUtils.consumerProps("groupid", "true", embeddedKafkaBroker);
          consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
          Consumer&lt;String, GenericRecord&gt; consumer = new DefaultKafkaConsumerFactory&lt;String, GenericRecord&gt;(consumerProps).createConsumer();
          consumer.subscribe(Collections.singleton(TEST_TOPIC));
          return consumer;
        }
  
        private Producer&lt;String, GenericRecord&gt; configureProducer() {
          Map&lt;String, Object&gt; producerProps = new HashMap&lt;&gt;(KafkaTestUtils.producerProps(embeddedKafkaBroker));
          producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
          producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
          producerProps.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, mockSchemaRegistryClient);
          producerProps.put(KafkaAvroSerializerConfig.AUTO_REGISTER_SCHEMAS, "false");
          return new DefaultKafkaProducerFactory&lt;String, GenericRecord&gt;(producerProps).createProducer();
        }
  
  }
  </code></pre>
  
  <h1>Error</h1>
  
  <pre><code>component.com.ms.listener.ConsumeKafkaMessageTest &gt; consumeKafkaMessage_receive_sucess() FAILED
      org.apache.kafka.common.KafkaException: Failed to construct kafka producer
          at org.apache.kafka.clients.producer.KafkaProducer.&lt;init&gt;(KafkaProducer.java:457)
          at org.apache.kafka.clients.producer.KafkaProducer.&lt;init&gt;(KafkaProducer.java:289)
          at org.springframework.kafka.core.DefaultKafkaProducerFactory.createKafkaProducer(DefaultKafkaProducerFactory.java:318)
          at org.springframework.kafka.core.DefaultKafkaProducerFactory.createProducer(DefaultKafkaProducerFactory.java:305)
          at component.com.ms.listener.ConsumeKafkaMessageTest.configureProducer(ConsumeKafkaMessageTest.java:125)
          at component.com.ms.listener.ConsumeKafkaMessageTest.consumeKafkaMessage_receive_sucess(ConsumeKafkaMessageTest.java:97)
  
          Caused by:
          io.confluent.common.config.ConfigException: Invalid value io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient@20751870 for configuration schema.registry.url: Expected a comma separated list.
              at io.confluent.common.config.ConfigDef.parseType(ConfigDef.java:345)
              at io.confluent.common.config.ConfigDef.parse(ConfigDef.java:249)
              at io.confluent.common.config.AbstractConfig.&lt;init&gt;(AbstractConfig.java:78)
              at io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig.&lt;init&gt;(AbstractKafkaAvroSerDeConfig.java:105)
              at io.confluent.kafka.serializers.KafkaAvroSerializerConfig.&lt;init&gt;(KafkaAvroSerializerConfig.java:32)
              at io.confluent.kafka.serializers.KafkaAvroSerializer.configure(KafkaAvroSerializer.java:48)
              at org.apache.kafka.common.serialization.ExtendedSerializer$Wrapper.configure(ExtendedSerializer.java:60)
              at org.apache.kafka.clients.producer.KafkaProducer.&lt;init&gt;(KafkaProducer.java:372)
              ... 5 more
  </code></pre>