Skip to content

Added a way to create topics and brokers by passing a custom configuration #15

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
May 17, 2016
Merged

Added a way to create topics and brokers by passing a custom configuration #15

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8986dc4
Added a way to create topics and brokers by passing a custom configur…
lucaqz May 15, 2016
5bcf9e1
moved the local vals to the main trait to avoid multiple instantiatio…
lucaqz May 16, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,11 @@ import java.net.InetSocketAddress
import java.util.Properties
import java.util.concurrent.Executors

import kafka.admin.AdminUtils
import kafka.consumer.{Consumer, ConsumerConfig, Whitelist}
import kafka.serializer.{Decoder, StringDecoder}
import kafka.server.{KafkaConfig, KafkaServer}
import kafka.utils.ZkUtils
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{Serializer, StringSerializer}
import org.apache.zookeeper.server.{ServerCnxnFactory, ZooKeeperServer}
Expand Down Expand Up @@ -75,6 +77,10 @@ sealed trait EmbeddedKafkaSupport {
val executorService = Executors.newFixedThreadPool(2)
implicit val executionContext = ExecutionContext.fromExecutorService(executorService)

val zkSessionTimeoutMs = 10000
val zkConnectionTimeoutMs = 10000
val zkSecurityEnabled = false

/**
* Starts a ZooKeeper instance and a Kafka broker, then executes the body passed as a parameter.
*
Expand Down Expand Up @@ -226,6 +232,7 @@ sealed trait EmbeddedKafkaSupport {
val zkAddress = s"localhost:${config.zooKeeperPort}"

val properties: Properties = new Properties
config.customBrokerProperties.foreach { case (key, value) => properties.setProperty(key, value) }
properties.setProperty("zookeeper.connect", zkAddress)
properties.setProperty("broker.id", "0")
properties.setProperty("host.name", "localhost")
Expand All @@ -239,4 +246,8 @@ sealed trait EmbeddedKafkaSupport {
broker.startup()
broker
}

def createCustomTopic(topic: String, topicConfig: Properties)(implicit config: EmbeddedKafkaConfig): Unit = {
Copy link

@chris-zen chris-zen May 18, 2016 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Working with Properties in Scala is a bit ugly, have you considered using a Map[String, String] for the interface of this function instead ?

AdminUtils.createTopic(ZkUtils(s"localhost:${config.zooKeeperPort}", zkSessionTimeoutMs, zkConnectionTimeoutMs, zkSecurityEnabled), topic, 1, 1, topicConfig)
}
}
2 changes: 1 addition & 1 deletion src/main/scala/net/manub/embeddedkafka/EmbeddedKafkaConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
package net.manub.embeddedkafka

case class EmbeddedKafkaConfig(kafkaPort: Int = 6001, zooKeeperPort: Int = 6000)
case class EmbeddedKafkaConfig(kafkaPort: Int = 6001, zooKeeperPort: Int = 6000, customBrokerProperties: Map[String, String] = Map.empty)

object EmbeddedKafkaConfig {
implicit val defaultConfig = EmbeddedKafkaConfig()
Expand Down
23 changes: 23 additions & 0 deletions src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,10 @@ package net.manub.embeddedkafka
import java.util.Properties
import java.util.concurrent.TimeoutException

import kafka.admin.AdminUtils
import kafka.consumer.{Consumer, ConsumerConfig, Whitelist}
import kafka.serializer.StringDecoder
import kafka.utils.ZkUtils
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.{ByteArraySerializer, StringSerializer}
import org.scalatest.exceptions.TestFailedException
Expand Down Expand Up @@ -114,6 +116,27 @@ class EmbeddedKafkaSpec extends EmbeddedKafkaSpecSupport with EmbeddedKafka {
}
}

"the createCustomTopic method" should {
"create a topic with a custom configuration" in {
implicit val config = EmbeddedKafkaConfig(customBrokerProperties = Map("log.cleaner.dedupe.buffer.size" -> "2000000"))
val topic = "test_custom_topic"

withRunningKafka {
val properties: Properties = new Properties
properties.put("cleanup.policy", "compact")

createCustomTopic(topic, properties)

val zkSessionTimeoutMs = 10000
val zkConnectionTimeoutMs = 10000
val zkSecurityEnabled = false

AdminUtils.topicExists(ZkUtils(s"localhost:${config.zooKeeperPort}", zkSessionTimeoutMs, zkConnectionTimeoutMs, zkSecurityEnabled), topic) shouldBe true

}
}
}

"the consumeFirstStringMessageFrom method" should {
"return a message published to a topic" in {
withRunningKafka {
Expand Down