title: Spring Boot Embedded Kafka can't connect
body: <p>I'm trying to write an integration test for my Kafka consumer. I've followed through with the <a href="https://docs.spring.io/spring-kafka/docs/2.2.4.RELEASE/reference/#reference" rel="noreferrer">official reference documentation</a> but when I start my test I only see this repeated ad infinitum:</p>
  
  <blockquote>
    <p>-2019-04-03 15:47:34.002 WARN 13120 --- [           main] org.apache.kafka.clients.NetworkClient   : [Consumer
    clientId=consumer-1, groupId=my-group] Connection to node -1 could not
    be established. Broker may not be available.</p>
  </blockquote>
  
  <p>What am I doing wrong?</p>
  
  <p>I'm using JUnit5, Spring Boot and <code>spring-kafka</code> and <code>spring-kafka-test</code>.</p>
  
  <p>I have the <code>@EnableKafka</code> annotation on my <code>@Configuration</code> class.</p>
  
  <p>This is how my test class looks like:</p>
  
  <pre><code>@ExtendWith(SpringExtension::class)
  @SpringBootTest(classes = [TestKafkaConfig::class])
  @DirtiesContext
  @EmbeddedKafka(
          partitions = 1,
          topics = [KafkaIntegrationTest.MY_TOPIC])
  class KafkaIntegrationTest {
  
      @Autowired
      private lateinit var embeddedKafka: EmbeddedKafkaBroker
  
      @Test
      fun test() {
          val senderProps = KafkaTestUtils.senderProps(embeddedKafka.brokersAsString)
          val template = KafkaTemplate(DefaultKafkaProducerFactory&lt;Int, String&gt;(senderProps))
          template.defaultTopic = KafkaIntegrationTest.MY_TOPIC
          template.sendDefault("foo")
      }
  }
  </code></pre>
  
  <p>my <code>application.yml</code> looks like this:</p>
  
  <pre><code>kafka:
    consumer:
      group-id: my-group
      bootstrap-servers: ${BOOTSTRAP_SERVERS:localhost:9092}
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        schema.registry.url: ${SCHEMA_REGISTRY_URL:http://localhost:8081}
        specific.avro.reader: true
  </code></pre>
  
  <p>I also tried setting up a <code>MockSchemaRegistryClient</code> but I get the exact same repeated message. (This is how I tried to set up the <code>MockSchemaRegistryClient</code>):</p>
  
  <pre><code>@TestConfiguration
  @Import(TestConfig::class)
  class TestKafkaConfig {
  
      @Autowired
      private lateinit var props: KafkaProperties
  
      @Bean
      fun schemaRegistryClient() = MockSchemaRegistryClient()
  
      @Bean
      fun kafkaAvroSerializer() = KafkaAvroSerializer(schemaRegistryClient())
  
      @Bean
      fun kafkaAvroDeserializer() = KafkaAvroDeserializer(schemaRegistryClient(), props.buildConsumerProperties())
  
      @Bean
      fun producerFactory(): ProducerFactory&lt;*, *&gt; = DefaultKafkaProducerFactory(
              props.buildProducerProperties(),
              StringSerializer(),
              kafkaAvroSerializer())
  
      @Bean
      fun consumerFactory(): ConsumerFactory&lt;*, *&gt; = DefaultKafkaConsumerFactory(
              props.buildConsumerProperties(),
              StringDeserializer(),
              kafkaAvroDeserializer()
      )
  
      @Bean
      fun kafkaListenerContainerFactory() = ConcurrentKafkaListenerContainerFactory&lt;Any, Any&gt;().apply {
          setConsumerFactory(consumerFactory() as ConsumerFactory&lt;in Any, in Any&gt;?)
      }
  
  }
  </code></pre>
  
  <p>What am I doing wrong?
  <strong>Note that</strong> I'm using the Confluent Schema Registry and try to deserialize from Avro.</p>
  
  <p>What I'm trying to test is whether my consumer works or not which looks like this:</p>
  
  <pre><code>open class SomeConsumer(private val someUseCase) {
  
      @KafkaListener(topics = ["\${kafka.some-topic}"])
      open fun processMessage(record: ConsumerRecord&lt;String, SomeObject&gt;) {
          someUseCase.call(record)
      }
  }
  </code></pre>