mock kafka embedded server for testing

build.gradle

@@ -60,6 +60,10 @@ dependencies {
     compile "org.jetbrains.kotlin:kotlin-reflect:1.3.71"
     //compile 'com.beust:klaxon:5.0.5'
     testCompile("org.junit.jupiter:junit-jupiter:5.4.2")
+
+    testImplementation "org.apache.kafka:kafka-clients:$kafkaV:test"
+    testImplementation "org.apache.kafka:kafka_2.11:$kafkaV"
+    testImplementation "org.apache.kafka:kafka_2.11:$kafkaV:test"
 }
 
 compileKotlin {

gradle/wrapper/gradle-wrapper.properties

-#Wed Oct 30 13:47:48 CET 2019
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.2.2-bin.zip
+#Fri Apr 03 11:36:55 CEST 2020
+distributionUrl=https\://services.gradle.org/distributions/gradle-6.2.2-all.zip
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStorePath=wrapper/dists

src/test/kotlin/KafkaEmbedded.kt

+/*
+ * sftp-reader
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package org.memobase
+
+import java.io.Closeable
+import java.nio.file.Files
+import java.util.Properties
+import kafka.admin.AdminUtils
+import kafka.admin.RackAwareMode
+import kafka.server.KafkaConfig
+import kafka.server.KafkaServer
+import kafka.utils.TestUtils
+import kafka.utils.ZkUtils
+import kafka.utils.`ZKStringSerializer$`
+import kafka.zk.EmbeddedZookeeper
+import org.I0Itec.zkclient.ZkClient
+import org.apache.kafka.common.utils.Time
+import scala.Option
+import scala.collection.JavaConversions
+
+class KafkaEmbedded(port: Int, topic: String) : Closeable {
+
+    private val server: KafkaServer
+    private val zkClient: ZkClient
+    private val zkServer: EmbeddedZookeeper = EmbeddedZookeeper()
+
+    init {
+        val zkConnect = "127.0.0.1:${zkServer.port()}"
+
+        val props = Properties()
+        props.setProperty("zookeeper.connect", zkConnect)
+        props.setProperty("broker.id", "0")
+        props.setProperty("log.dirs", Files.createTempDirectory("kafka-").toAbsolutePath().toString())
+        props.setProperty("listeners", "PLAINTEXT://127.0.0.1:$port")
+        props.setProperty("offsets.topic.replication.factor", "1")
+
+        server = KafkaServer(KafkaConfig(props), Time.SYSTEM, Option.apply("kafka-broker"), JavaConversions.asScalaBuffer(emptyList()))
+        server.startup()
+
+        zkClient = ZkClient(zkConnect, 30000, 30000, `ZKStringSerializer$`.`MODULE$`)
+        val zkUtils = ZkUtils.apply(zkClient, false)
+        AdminUtils.createTopic(zkUtils, topic, 1, 1, Properties(), RackAwareMode.`Disabled$`.`MODULE$`)
+
+        TestUtils.waitUntilMetadataIsPropagated(scala.collection.JavaConversions.asScalaBuffer(listOf(server)), topic, 0, 5000)
+    }
+
+    override fun close() {
+        server.shutdown()
+        server.awaitShutdown()
+        zkClient.close()
+        zkServer.shutdown()
+    }
+}

src/test/kotlin/Tests.kt

@@ -15,7 +15,7 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
-package org.swissbib.linked
+package org.memobase
 
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
@@ -25,9 +25,12 @@ class Tests {
 
     @Test
     fun testCase1() {
-        assertAll("test case 1",
-            { assertEquals("null", "null") },
-            { assertEquals("null", "null") }
-        )
+        KafkaEmbedded(12301, "test").use {
+
+            assertAll("test case 1",
+                { assertEquals("null", "null") },
+                { assertEquals("null", "null") }
+            )
+        }
     }
 }