scalatest-embedded-kafka

scalatest-embedded-kafka copied to clipboard

I started working PRs for #109 and #92 but have run into a problem running tests.

I'm posting here in hopes of getting some help in resolving the issue.

So far I've tried:

• downgrading the version of Zookeeper
• upgrading my version of Java
• randomising the ports assigned in EmbeddedKafkaObjectSpec

I came across ZOOKEEPER-1441 which outlines my issue but has been left open.

I am able to run tests individually but multiple tests, even using separate ports, result in the second test and onwards failing for the following reason.

Address already in use
java.net.BindException: Address already in use
	at sun.nio.ch.Net.bind0(Native Method)
	at sun.nio.ch.Net.bind(Net.java:433)
	at sun.nio.ch.Net.bind(Net.java:425)
	at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)
	at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)
	at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:67)
	at org.apache.zookeeper.server.NIOServerCnxnFactory.configure(NIOServerCnxnFactory.java:90)
	at net.manub.embeddedkafka.EmbeddedKafkaSupport.startZooKeeper(EmbeddedKafka.scala:587)
	at net.manub.embeddedkafka.EmbeddedKafkaSupport.startZooKeeper$(EmbeddedKafka.scala:578)
	at net.manub.embeddedkafka.EmbeddedKafka$.startZooKeeper(EmbeddedKafka.scala:29)
	at net.manub.embeddedkafka.EmbeddedKafka$.start(EmbeddedKafka.scala:43)
	at net.manub.embeddedkafka.EmbeddedKafkaObjectSpec$$anon$3.<init>(EmbeddedKafkaObjectSpec.scala:38)
	at net.manub.embeddedkafka.EmbeddedKafkaObjectSpec.$anonfun$new$5(EmbeddedKafkaObjectSpec.scala:37)
	at org.scalatest.OutcomeOf.outcomeOf(OutcomeOf.scala:85)
	at org.scalatest.OutcomeOf.outcomeOf$(OutcomeOf.scala:83)
	at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
	at org.scalatest.Transformer.apply(Transformer.scala:22)
	at org.scalatest.Transformer.apply(Transformer.scala:20)
	at org.scalatest.WordSpecLike$$anon$1.apply(WordSpecLike.scala:1078)
	at org.scalatest.TestSuite.withFixture(TestSuite.scala:196)
	at org.scalatest.TestSuite.withFixture$(TestSuite.scala:195)
	at net.manub.embeddedkafka.EmbeddedKafkaSpecSupport.withFixture(embeddedKafkaSpecSupport.scala:22)
	at org.scalatest.WordSpecLike.invokeWithFixture$1(WordSpecLike.scala:1076)
	at org.scalatest.WordSpecLike.$anonfun$runTest$1(WordSpecLike.scala:1088)
	at org.scalatest.SuperEngine.runTestImpl(Engine.scala:289)
	at org.scalatest.WordSpecLike.runTest(WordSpecLike.scala:1088)
	at org.scalatest.WordSpecLike.runTest$(WordSpecLike.scala:1070)
	at net.manub.embeddedkafka.EmbeddedKafkaSpecSupport.runTest(embeddedKafkaSpecSupport.scala:22)
	at org.scalatest.WordSpecLike.$anonfun$runTests$1(WordSpecLike.scala:1147)
	at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:396)
	at scala.collection.immutable.List.foreach(List.scala:389)
	at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384)
	at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:373)
	at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:410)
	at scala.collection.immutable.List.foreach(List.scala:389)
	at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384)
	at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:373)
	at org.scalatest.SuperEngine.$anonfun$runTestsInBranch$1(Engine.scala:410)
	at scala.collection.immutable.List.foreach(List.scala:389)
	at org.scalatest.SuperEngine.traverseSubNodes$1(Engine.scala:384)
	at org.scalatest.SuperEngine.runTestsInBranch(Engine.scala:379)
	at org.scalatest.SuperEngine.runTestsImpl(Engine.scala:461)
	at org.scalatest.WordSpecLike.runTests(WordSpecLike.scala:1147)
	at org.scalatest.WordSpecLike.runTests$(WordSpecLike.scala:1146)
	at net.manub.embeddedkafka.EmbeddedKafkaSpecSupport.runTests(embeddedKafkaSpecSupport.scala:22)
	at org.scalatest.Suite.run(Suite.scala:1147)
	at org.scalatest.Suite.run$(Suite.scala:1129)
	at net.manub.embeddedkafka.EmbeddedKafkaSpecSupport.org$scalatest$WordSpecLike$$super$run(embeddedKafkaSpecSupport.scala:22)
	at org.scalatest.WordSpecLike.$anonfun$run$1(WordSpecLike.scala:1192)
	at org.scalatest.SuperEngine.runImpl(Engine.scala:521)
	at org.scalatest.WordSpecLike.run(WordSpecLike.scala:1192)
	at org.scalatest.WordSpecLike.run$(WordSpecLike.scala:1190)
	at net.manub.embeddedkafka.EmbeddedKafkaSpecSupport.org$scalatest$BeforeAndAfterAll$$super$run(embeddedKafkaSpecSupport.scala:22)
	at org.scalatest.BeforeAndAfterAll.liftedTree1$1(BeforeAndAfterAll.scala:213)
	at org.scalatest.BeforeAndAfterAll.run(BeforeAndAfterAll.scala:210)
	at org.scalatest.BeforeAndAfterAll.run$(BeforeAndAfterAll.scala:208)
	at net.manub.embeddedkafka.EmbeddedKafkaSpecSupport.run(embeddedKafkaSpecSupport.scala:22)
	at org.scalatest.tools.SuiteRunner.run(SuiteRunner.scala:45)
	at org.scalatest.tools.Runner$.$anonfun$doRunRunRunDaDoRunRun$13(Runner.scala:1340)
	at org.scalatest.tools.Runner$.$anonfun$doRunRunRunDaDoRunRun$13$adapted(Runner.scala:1334)
	at scala.collection.immutable.List.foreach(List.scala:389)
	at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:1334)
	at org.scalatest.tools.Runner$.$anonfun$runOptionallyWithPassFailReporter$24(Runner.scala:1031)
	at org.scalatest.tools.Runner$.$anonfun$runOptionallyWithPassFailReporter$24$adapted(Runner.scala:1010)
	at org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:1500)
	at org.scalatest.tools.Runner$.runOptionallyWithPassFailReporter(Runner.scala:1010)
	at org.scalatest.tools.Runner$.run(Runner.scala:850)
	at org.scalatest.tools.Runner.run(Runner.scala)
	at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.runScalaTest2(ScalaTestRunner.java:131)
	at org.jetbrains.plugins.scala.testingSupport.scalaTest.ScalaTestRunner.main(ScalaTestRunner.java:28)

Any thoughts would be appreicated.

I've seen this issue on multiple Mac's and running this project's Travis build.

Sorry to reply to this just now.

Are you allowing parallel execution of your tests?

Rerunning the latest version of the project, I still get the same "Address already in use error" whilst running sbt test.

The only references to parallel execution I can see are in build.sbt:

parallelExecution in ThisBuild := false and parallelExecution in Test := false

I'm sorry about this - I haven't been able to reproduce. What's the frequency of this failure?

Could you share some of the test code you're talking about?

I wonder if this could actually be related to #150.

Any chance you can try with the current master? If this is fixed I can release 1.1.1.1.

Unfortunately I still do. I'll spend some time looking further into why. The fact that this issue didn't resolve it gives me something to go on.

java.net.BindException: Address already in use: bind at sun.nio.ch.Net.bind0(Native Method) at sun.nio.ch.Net.bind(Net.java:433) at sun.nio.ch.Net.bind(Net.java:425) at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223) at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74) at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:67) at org.apache.zookeeper.server.NIOServerCnxnFactory.configure(NIOServerCnxnFactory.java:90) at org.apache.zookeeper.server.ZooKeeperServerMain.runFromConfig(ZooKeeperServerMain.java:120) at org.apache.zookeeper.server.ZooKeeperServerMain.initializeAndRun(ZooKeeperServerMain.java:89) at org.apache.zookeeper.server.ZooKeeperServerMain.main(ZooKeeperServerMain.java:55) at org.apache.zookeeper.server.quorum.QuorumPeerMain.initializeAndRun(QuorumPeerMain.java:119) at org.apache.zookeeper.server.quorum.QuorumPeerMain.main(QuorumPeerMain.java:81)

My error when starting zookeeper, any solution for this ??

This error is also happening to me when I use withRunningKafka on different test files.

Ex: I have a testSpec:

class TopicCreationSpec extends UnitSpec with EmbeddedKafka {
 it should "Create input kafka topic when input is from kafka" in {
    withRunningKafka {
      val topicName = "input_test_topic"
      assert(!KafkaUtils.getTopics.contains(topicName))
      val jsonNode = configValidator.check(baseYaml).right.get
      jsonNode.input.maybeCreateTopic

      KafkaUtils.waitForTopicCreation(topicName)

      assert(KafkaUtils.getTopics.contains(topicName))
    }
  }
}

And I have other testSpec:

class InputSpec  extends UnitSpec with EmbeddedKafka {

  it should "Create input kafka topic when input is from kafka" in {
    withRunningKafka {

      val topicName = "input_test_topic"
      assert(!KafkaUtils.getTopics.contains(topicName))
      val jsonNode = configValidator.check(baseYaml).right.get
      jsonNode.input.maybeCreateTopic

      KafkaUtils.waitForTopicCreation(topicName)

      assert(KafkaUtils.getTopics.contains(topicName))
    }
  }
}

And it fails with

InputSpec: [info] - should Create input kafka topic when input is from kafka *** FAILED *** [info] java.net.BindException: Address already in use [info] at sun.nio.ch.Net.bind0(Native Method) [info] at sun.nio.ch.Net.bind(Net.java:433) [info] at sun.nio.ch.Net.bind(Net.java:425) [info] at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223) [info] at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74) [info] at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:67) [info] at org.apache.zookeeper.server.NIOServerCnxnFactory.configure(NIOServerCnxnFactory.java:687) [info] at org.apache.zookeeper.server.ServerCnxnFactory.configure(ServerCnxnFactory.java:76) [info] at net.manub.embeddedkafka.ops.ZooKeeperOps$class.startZooKeeper(zooKeeperOps.scala:27) [info] at net.manub.embeddedkafka.EmbeddedKafka$.startZooKeeper(EmbeddedKafka.scala:50) [info] ...

I'm on macOS Catalina. If I just move all the tests to the same file everything is fine and working.