issue #221 allow more than 1 URI to be given to connect serverUri, re… (#226)

* issue #221 allow more than 1 URI to be given to connect serverUri, resulting in a routingDriver

* PR feedback from @jexp; update driver, use improved addr approach

Author: moxious

kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jService.kt

@@ -14,6 +14,7 @@ import org.neo4j.driver.v1.Driver
 import org.neo4j.driver.v1.GraphDatabase
 import org.neo4j.driver.v1.exceptions.ClientException
 import org.neo4j.driver.v1.exceptions.TransientException
+import org.neo4j.driver.v1.net.ServerAddress
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import streams.kafka.connect.sink.converters.Neo4jValueConverter
@@ -66,8 +67,10 @@ class Neo4jService(private val config: Neo4jSinkConnectorConfig):
         configBuilder.withConnectionAcquisitionTimeout(this.config.connectionAcquisitionTimeout, TimeUnit.MILLISECONDS)
         configBuilder.withLoadBalancingStrategy(this.config.loadBalancingStrategy)
         configBuilder.withMaxTransactionRetryTime(config.retryBackoff, TimeUnit.MILLISECONDS)
+        configBuilder.withResolver { address -> this.config.serverUri.map { ServerAddress.of(it.host, it.port) }.toSet() }
         val neo4jConfig = configBuilder.toConfig()
-        this.driver = GraphDatabase.driver(this.config.serverUri, authToken, neo4jConfig)
+
+        this.driver = GraphDatabase.driver(this.config.serverUri.get(0), authToken, neo4jConfig)
     }
 
     fun close() {

kafka-connect-neo4j/src/main/kotlin/streams/kafka/connect/sink/Neo4jSinkConnectorConfig.kt

@@ -19,6 +19,7 @@ import streams.service.Topics
 import streams.service.sink.strategy.SourceIdIngestionStrategyConfig
 import java.io.File
 import java.net.URI
+import java.net.URISyntaxException
 import java.util.concurrent.TimeUnit
 
 enum class AuthenticationType {
@@ -46,7 +47,7 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
     val authenticationRealm: String
     val authenticationKerberosTicket: String
 
-    val serverUri: URI
+    val serverUri: List<URI>
     val connectionMaxConnectionLifetime: Long
     val connectionLifenessCheckTimeout: Long
     val connectionPoolMaxSize: Int
@@ -85,7 +86,7 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
         authenticationPassword = getPassword(AUTHENTICATION_BASIC_PASSWORD).value()
         authenticationKerberosTicket = getPassword(AUTHENTICATION_KERBEROS_TICKET).value()
 
-        serverUri = ConfigUtils.uri(this, SERVER_URI)
+        serverUri = listOfURIs(SERVER_URI, getString(SERVER_URI))
         connectionLifenessCheckTimeout = getLong(CONNECTION_LIVENESS_CHECK_TIMEOUT_MSECS)
         connectionMaxConnectionLifetime = getLong(CONNECTION_MAX_CONNECTION_LIFETIME_MSECS)
         connectionPoolMaxSize = getInt(CONNECTION_POOL_MAX_SIZE)
@@ -106,6 +107,10 @@ class Neo4jSinkConnectorConfig(originals: Map<*, *>): AbstractConfig(config(), o
         validateAllTopics(originals)
     }
 
+    private fun listOfURIs(key: String, value: String): List<URI> {
+        return value.split(",").map { URI(it) }
+    }
+
     private fun validateAllTopics(originals: Map<*, *>) {
         TopicUtils.validate<ConfigException>(this.topics)
         val topics = if (originals.containsKey(SinkTask.TOPICS_CONFIG)) {

kafka-connect-neo4j/src/test/kotlin/streams/kafka/connect/sink/Neo4jSinkConnectorConfigTest.kt

@@ -39,17 +39,21 @@ class Neo4jSinkConnectorConfigTest {
 
     @Test
     fun `should return the configuration`() {
+        val a = "bolt://neo4j:7687"
+        val b = "bolt://neo4j2:7687"
+
         val originals = mapOf(SinkConnector.TOPICS_CONFIG to "foo",
                 "${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}foo" to "CREATE (p:Person{name: event.firstName})",
-                Neo4jSinkConnectorConfig.SERVER_URI to "bolt://neo4j:7687",
+                Neo4jSinkConnectorConfig.SERVER_URI to "$a,$b", // Check for string trimming
                 Neo4jSinkConnectorConfig.BATCH_SIZE to 10,
                 Neo4jSinkConnectorConfig.AUTHENTICATION_BASIC_USERNAME to "FOO",
                 Neo4jSinkConnectorConfig.AUTHENTICATION_BASIC_PASSWORD to "BAR")
         val config = Neo4jSinkConnectorConfig(originals)
 
         assertEquals(originals["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}foo"], config.topics.cypherTopics["foo"])
         assertFalse { config.encryptionEnabled }
-        assertEquals(originals[Neo4jSinkConnectorConfig.SERVER_URI], config.serverUri.toString())
+        assertEquals(a, config.serverUri.get(0).toString())
+        assertEquals(b, config.serverUri.get(1).toString())
         assertEquals(originals[Neo4jSinkConnectorConfig.BATCH_SIZE], config.batchSize)
         assertEquals(Config.TrustStrategy.Strategy.TRUST_ALL_CERTIFICATES, config.encryptionTrustStrategy)
         assertEquals(AuthenticationType.BASIC, config.authenticationType)
@@ -67,6 +71,22 @@ class Neo4jSinkConnectorConfigTest {
         assertEquals(Neo4jSinkConnectorConfig.BATCH_TIMEOUT_DEFAULT, config.batchTimeout)
     }
 
+    @Test
+    fun `should return valid configuration with multiple URIs`() {
+        val a = "bolt://neo4j:7687"
+        val b = "bolt://neo4j2:7687"
+        val c = "bolt://neo4j3:7777"
+
+        val originals = mapOf(SinkConnector.TOPICS_CONFIG to "foo",
+                "${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}foo" to "CREATE (p:Person{name: event.firstName})",
+                Neo4jSinkConnectorConfig.SERVER_URI to "$a,$b,$c")
+        val config = Neo4jSinkConnectorConfig(originals)
+
+        assertEquals(a, config.serverUri.get(0).toString())
+        assertEquals(b, config.serverUri.get(1).toString())
+        assertEquals(c, config.serverUri.get(2).toString())
+    }
+
     @Test
     fun `should return the configuration with shuffled topic order`() {
         val originals = mapOf(SinkConnector.TOPICS_CONFIG to "bar,foo",
@@ -80,7 +100,7 @@ class Neo4jSinkConnectorConfigTest {
 
         assertEquals(originals["${Neo4jSinkConnectorConfig.TOPIC_CYPHER_PREFIX}foo"], config.topics.cypherTopics["foo"])
         assertFalse { config.encryptionEnabled }
-        assertEquals(originals[Neo4jSinkConnectorConfig.SERVER_URI], config.serverUri.toString())
+        assertEquals(originals[Neo4jSinkConnectorConfig.SERVER_URI], config.serverUri.get(0).toString())
         assertEquals(originals[Neo4jSinkConnectorConfig.BATCH_SIZE], config.batchSize)
         assertEquals(Config.TrustStrategy.Strategy.TRUST_ALL_CERTIFICATES, config.encryptionTrustStrategy)
         assertEquals(AuthenticationType.BASIC, config.authenticationType)

pom.xml

@@ -39,7 +39,7 @@
         <kafka.version>1.0.1</kafka.version>
         <jackson.version>2.9.7</jackson.version>
         <kotlin.compiler.incremental>true</kotlin.compiler.incremental>
-        <neo4j.java.driver.version>1.6.3</neo4j.java.driver.version>
+        <neo4j.java.driver.version>1.7.5</neo4j.java.driver.version>
         <testcontainers.version>1.9.1</testcontainers.version>
     </properties>