Problems running embedded kafka with multiple scalatests

[IoDmitri]

Im having issues with embedded kafka when I run it within multiple scalatesst suits. Here is my SO question that displays more of the code and details the issue.

[manub]

I replied to the SO question - you should probably look into EmbeddedKafka.start() and EmbeddedKafka.stop() on beforeAll and afterAll.

[blbradley]

@IoDmitri There are certain situations when using this library that could cause your tests to fail because of SBT's parallel execution of tests.

I've found that disabling parallel execution helps with those situations: parallelExecution in Test := false.

[manub]

@IoDmitri any news on this?

[IoDmitri]

@manub The issue has been resolved by running before and after each, this prevents the shutdown/startup from bleeding into other test suits.

[manub]

Brilliant!

[prayagupa]

@manub
Strange, adding start and stop beforeEach and afterEach does not work for me. I also turned off sbt parallelExecution

class KafkaEventConsumerSpecs extends FunSuite with BeforeAndAfterEach {

  implicit val streamingConfig = EmbeddedKafkaConfig(kafkaPort = 9092, zooKeeperPort = 2181)

  override protected def beforeEach(): Unit = {
    EmbeddedKafka.start()
  }

  override protected def afterEach(): Unit = {
    EmbeddedKafka.stop()
  }

 test("consumes 1") {}
 test("consumes 2") {}
}

sbt

name := "es-driver"

version := "1.0"

scalaVersion := "2.11.8"

parallelExecution in Test := false

In above example, the topics in first test are visible to second test.

[manub]

@prayagupd which version of scalatest-embedded-kafka are you using? You may be hit by a bug I fixed in ca287be.

[prayagupa]

@manub I'm using 0.11.0 (https://github.com/nihil-os/streaming-driver/blob/master/build.sbt#L29). The issue actually is while running multiple tests at the same time. If I run individually works fine.

Saw

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=testKafkaEventConsumer
	at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:437)
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1898)
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:966)
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:900)
	at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:324)
	at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522)
	at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:58)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:695)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:584)
	at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:566)

Seems first test did not clean up the consumer info, as the second test has similar group.id and consumer.id, it inherits consumer position from from test.

[manub]

It seems to me you're registering twice the same consumerId and Kafka doesn't allow that. If you're running tests in parallel you may want to change consumerId for every test (and potentially groupId to avoid registering the two tests in the same group which will lead to unexpected behaviour).