forked from neo4j-contrib/neo4j-streams
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fixes neo4j-contrib#181: Validate configuration before attempting to …
…start
- Loading branch information
Showing
9 changed files
with
200 additions
and
42 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,22 +1,31 @@ | ||
| package streams.utils | ||
|
|
||
| import java.io.IOException | ||
| import java.net.Socket | ||
| import java.net.URI | ||
|
|
||
| object ValidationUtils { | ||
|
|
||
| fun validateTopics(cdcMergeTopics: Set<String>, cdcSchemaTopics: Set<String>, | ||
| cypherTopics: Set<String>, nodePatternTopics: Set<String>, relPatternTopics: Set<String>) { | ||
| val allTopicsLists = mutableListOf<String>() | ||
| allTopicsLists.addAll(cdcMergeTopics) | ||
| allTopicsLists.addAll(cdcSchemaTopics) | ||
| allTopicsLists.addAll(cypherTopics) | ||
| allTopicsLists.addAll(nodePatternTopics) | ||
| allTopicsLists.addAll(relPatternTopics) | ||
| val crossDefinedTopics = allTopicsLists.map { it to 1 } | ||
| .groupBy({ it.first }, { it.second }) | ||
| .mapValues { it.value.reduce { acc, i -> acc + i } } | ||
| .filterValues { it > 1 } | ||
| .keys | ||
| if (crossDefinedTopics.isNotEmpty()) { | ||
| throw RuntimeException("The following topics are cross defined: $crossDefinedTopics") | ||
| } | ||
| fun isServerReachable(url: String, port: Int): Boolean = try { | ||
| Socket(url, port).use { true } | ||
| } catch (e: IOException) { | ||
| false | ||
| } | ||
|
|
||
| fun checkServersUnreachable(urls: String, separator: String = ","): List<String> = urls | ||
| .split(separator) | ||
| .map { | ||
| val uri = URI.create(it) | ||
| when (uri.host.isNullOrBlank()) { | ||
| true -> { | ||
| val splitted = it.split(":") | ||
| URI("fake-scheme", "", splitted.first(), splitted.last().toInt(), | ||
| "", "", "") | ||
| } | ||
| else -> uri | ||
| } | ||
| } | ||
| .filter { uri -> !isServerReachable(uri.host, uri.port) } | ||
| .map { if (it.scheme == "fake-scheme") "${it.host}:${it.port}" else it.toString() } | ||
|
|
||
| } |
36 changes: 36 additions & 0 deletions
36
common/src/test/kotlin/streams/utils/ValidationUtilsTest.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package streams.utils | ||
|
|
||
| import org.junit.Test | ||
| import org.testcontainers.containers.GenericContainer | ||
| import kotlin.test.assertEquals | ||
| import kotlin.test.assertTrue | ||
|
|
||
| class FakeWebServer: GenericContainer<FakeWebServer>("alpine") { | ||
| override fun start() { | ||
| this.withCommand("/bin/sh", "-c", "while true; do { echo -e 'HTTP/1.1 200 OK'; echo ; } | nc -l -p 8000; done") | ||
| .withExposedPorts(8000) | ||
| super.start() | ||
| } | ||
|
|
||
| fun getUrl() = "http://localhost:${getMappedPort(8000)}" | ||
| } | ||
|
|
||
| class ValidationUtilsTest { | ||
|
|
||
| @Test | ||
| fun `should reach the server`() { | ||
| val httpServer = FakeWebServer() | ||
| httpServer.start() | ||
| assertTrue { ValidationUtils.checkServersUnreachable(httpServer.getUrl()).isEmpty() } | ||
| httpServer.stop() | ||
| } | ||
|
|
||
| @Test | ||
| fun `should not reach the server`() { | ||
| val urls = "http://my.fake.host:1234,PLAINTEXT://my.fake.host1:1234,my.fake.host2:1234" | ||
| val checkServersUnreachable = ValidationUtils | ||
| .checkServersUnreachable(urls) | ||
| assertTrue { checkServersUnreachable.isNotEmpty() } | ||
| assertEquals(urls.split(",").toList(), checkServersUnreachable) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 55 additions & 0 deletions
55
consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkNoConfigurationIT.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| package integrations.kafka | ||
|
|
||
| import io.confluent.kafka.serializers.KafkaAvroDeserializer | ||
| import org.junit.Test | ||
| import org.neo4j.kernel.internal.GraphDatabaseAPI | ||
| import org.neo4j.test.TestGraphDatabaseFactory | ||
| import org.testcontainers.containers.GenericContainer | ||
| import kotlin.test.assertEquals | ||
|
|
||
|
|
||
| class FakeWebServer: GenericContainer<FakeWebServer>("alpine") { | ||
| override fun start() { | ||
| this.withCommand("/bin/sh", "-c", "while true; do { echo -e 'HTTP/1.1 200 OK'; echo ; } | nc -l -p 8000; done") | ||
| .withExposedPorts(8000) | ||
| super.start() | ||
| } | ||
|
|
||
| fun getUrl() = "http://localhost:${getMappedPort(8000)}" | ||
| } | ||
|
|
||
| class KafkaEventSinkNoConfigurationIT { | ||
|
|
||
| private val topic = "no-config" | ||
|
|
||
| @Test | ||
| fun `the db should start even with no bootstrap servers provided()`() { | ||
| val db = TestGraphDatabaseFactory() | ||
| .newImpermanentDatabaseBuilder() | ||
| .setConfig("kafka.bootstrap.servers", "") | ||
| .setConfig("streams.sink.enabled", "true") | ||
| .setConfig("streams.sink.topic.cypher.$topic", "CREATE (p:Place{name: event.name, coordinates: event.coordinates, citizens: event.citizens})") | ||
| .newGraphDatabase() as GraphDatabaseAPI | ||
| val count = db.execute("MATCH (n) RETURN COUNT(n) AS count").columnAs<Long>("count").next() | ||
| assertEquals(0L, count) | ||
| } | ||
|
|
||
| @Test | ||
| fun `the db should start even with AVRO serializers and no schema registry url provided()`() { | ||
| val fakeWebServer = FakeWebServer() | ||
| fakeWebServer.start() | ||
| val url = fakeWebServer.getUrl().replace("http://", "") | ||
| val db = TestGraphDatabaseFactory() | ||
| .newImpermanentDatabaseBuilder() | ||
| .setConfig("kafka.bootstrap.servers", url) | ||
| .setConfig("kafka.zookeeper.connect", url) | ||
| .setConfig("streams.sink.enabled", "true") | ||
| .setConfig("streams.sink.topic.cypher.$topic", "CREATE (p:Place{name: event.name, coordinates: event.coordinates, citizens: event.citizens})") | ||
| .setConfig("kafka.key.deserializer", KafkaAvroDeserializer::class.java.name) | ||
| .setConfig("kafka.value.deserializer", KafkaAvroDeserializer::class.java.name) | ||
| .newGraphDatabase() as GraphDatabaseAPI | ||
| val count = db.execute("MATCH (n) RETURN COUNT(n) AS count").columnAs<Long>("count").next() | ||
| assertEquals(0L, count) | ||
| fakeWebServer.stop() | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters