title: Using Kafka with Micronaut
body: <p>Are there any example projects showing how to use Kafka with Micronaut?  I am having problems with getting it to work.</p>
  
  <p>I have the following producer:</p>
  
  <pre><code>@KafkaClient
  interface AppClient {
  
  @Topic("topic-name")
      void sendMessage(@KafkaKey String id, Event event)
  }
  </code></pre>
  
  <p>and listener:</p>
  
  <pre><code>@KafkaListener(
      groupId="group-id",
      offsetReset = OffsetReset.EARLIEST
  )
  class AppListener {
  
  @Topic("topic-name")
      void onMessage(Event event) {
          // do stuff
      }
  }
  </code></pre>
  
  <p>My application.yml contains:</p>
  
  <pre><code>kafka:
    bootstrap:
      servers: localhost:2181
  </code></pre>
  
  <p>and application-test.yml (is this right and should it be in the same directory as application.yml?.  Also unsure how the embedded server should be used):</p>
  
  <pre><code>kafka:
    #  embedded:
    #    enabled: true
    #    topics: promo-api-promotions
    bootstrap:
      servers: localhost:9092
  </code></pre>
  
  <p>My test looks like:</p>
  
  <pre><code>@MicronautTest
  class AppSpec extends Specification {
  
  @Shared
  @AutoCleanup
  EmbeddedServer server = ApplicationContext.run(EmbeddedServer)
  
  @Shared
  private AppClient appClient =
          server.applicationContext.getBean(AppClient)
  
  def 'The upload endpoint is called'() {
    // test here
    appClient.sendMessage(id, event)
    // other test stuff
  }
  </code></pre>
  
  <p>The main problems I am having are:</p>
  
  <ol>
  <li><p>My consumer is not consuming from my topic.  I can see the producer creates the topic in Kafka and the client group is created, but the offset stays at 0.</p></li>
  <li><p>I am having problems when the test is started up where it looks as if two instances of the client are created and therefore the MBean registration fails (also, if I try to use the embedded Kafka, I get a different message about port 9092 already being in use because it tries to start the server up twice):</p>
  
  <p>javax.management.InstanceAlreadyExistsException: 
  kafka.consumer:type=app-info,id=app-kafka-client-app-listener
  at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437)
  at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898)</p></li>
  </ol>