feat: implement sink connector registration

Author: fbiville

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/EnvBackedSetting.kt

@@ -16,14 +16,15 @@
  */
 package org.neo4j.connectors.kafka.testing
 
+import org.neo4j.connectors.kafka.testing.WordSupport.camelCaseToUpperSnakeCase
 import org.neo4j.connectors.kafka.testing.source.DEFAULT_TO_ENV
 
 internal class EnvBackedSetting<T : Annotation>(
     private val name: String,
-    private val envVarName: String,
     private val getter: (T) -> String,
     private val envAccessor: (String) -> String? = System::getenv,
 ) {
+  private val envVarName = camelCaseToUpperSnakeCase(name)
 
   fun isValid(annotation: T): Boolean {
     return getter(annotation) != DEFAULT_TO_ENV || envAccessor(envVarName) != null

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/WordSupport.kt

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.connectors.kafka.testing
+
+import kotlin.math.absoluteValue
+
+object WordSupport {
+
+  // implementation note: this is a naive pluralization helper, which is not meant to work outside
+  // English
+  fun pluralize(count: Int, singular: String, plural: String): String {
+    if (count.absoluteValue < 2) {
+      return singular
+    }
+    return plural
+  }
+
+  fun camelCaseToUpperSnakeCase(word: String): String {
+    fun convertChar(char: Char): List<Char> {
+      return if (Character.isUpperCase(char)) {
+        listOf('_', char)
+      } else listOf(char.uppercaseChar())
+    }
+
+    return word.flatMap { convertChar(it) }.joinToString("")
+  }
+}

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/sink/Neo4jSink.kt

@@ -23,6 +23,7 @@ import org.neo4j.connectors.kafka.testing.source.DEFAULT_TO_ENV
 @Retention(AnnotationRetention.RUNTIME)
 @ExtendWith(Neo4jSinkExtension::class)
 annotation class Neo4jSink(
+    val kafkaConnectExternalUri: String = DEFAULT_TO_ENV,
     val neo4jUri: String = DEFAULT_TO_ENV,
     val neo4jUser: String = DEFAULT_TO_ENV,
     val neo4jPassword: String = DEFAULT_TO_ENV,

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/sink/Neo4jSinkExtension.kt

@@ -24,24 +24,27 @@ import org.junit.jupiter.api.extension.ExtensionConfigurationException
 import org.junit.jupiter.api.extension.ExtensionContext
 import org.neo4j.connectors.kafka.testing.AnnotationSupport
 import org.neo4j.connectors.kafka.testing.EnvBackedSetting
+import org.neo4j.connectors.kafka.testing.WordSupport.pluralize
 
-internal class Neo4jSinkExtension(
+/*nocommit*/ class Neo4jSinkExtension(
     // visible for testing
     envAccessor: (String) -> String? = System::getenv
 ) : ExecutionCondition, BeforeEachCallback, AfterEachCallback {
 
   private lateinit var sinkAnnotation: Neo4jSink
+  private lateinit var sink: Neo4jSinkRegistration
 
-  private val neo4jUri =
-      EnvBackedSetting<Neo4jSink>("neo4jUri", "NEO4J_URI", { it.neo4jUri }, envAccessor)
-  private val neo4jUser =
-      EnvBackedSetting<Neo4jSink>("neo4jUser", "NEO4J_USER", { it.neo4jUser }, envAccessor)
-  private val neo4jPassword =
+  private val kafkaConnectExternalUri =
       EnvBackedSetting<Neo4jSink>(
-          "neo4jPassword", "NEO4J_PASSWORD", { it.neo4jPassword }, envAccessor)
+          "kafkaConnectExternalUri", { it.kafkaConnectExternalUri }, envAccessor)
+  private val neo4jUri = EnvBackedSetting<Neo4jSink>("neo4jUri", { it.neo4jUri }, envAccessor)
+  private val neo4jUser = EnvBackedSetting<Neo4jSink>("neo4jUser", { it.neo4jUser }, envAccessor)
+  private val neo4jPassword =
+      EnvBackedSetting<Neo4jSink>("neo4jPassword", { it.neo4jPassword }, envAccessor)
 
   private val envSettings =
       listOf(
+          kafkaConnectExternalUri,
           neo4jUri,
           neo4jUser,
           neo4jPassword,
@@ -58,6 +61,11 @@ internal class Neo4jSinkExtension(
         errors.add(it.errorMessage())
       }
     }
+    val topicCount = metadata.topics.size
+    if (topicCount != metadata.queries.size) {
+      errors.add(
+          "Expected $topicCount ${pluralize(topicCount, "query", "queries")}, but got ${metadata.queries.size}. There must be as many topics as queries defined.")
+    }
     if (errors.isNotEmpty()) {
       throw ExtensionConfigurationException(
           "\nMissing settings, see details below:\n\t${errors.joinToString("\n\t")}",
@@ -68,11 +76,17 @@ internal class Neo4jSinkExtension(
     return ConditionEvaluationResult.enabled("@Neo4jSink and environment properly configured")
   }
 
-  override fun beforeEach(p0: ExtensionContext?) {
-    TODO("Not yet implemented")
+  override fun beforeEach(extensionContext: ExtensionContext?) {
+    sink =
+        Neo4jSinkRegistration(
+            neo4jUri = neo4jUri.read(sinkAnnotation),
+            neo4jUser = neo4jUser.read(sinkAnnotation),
+            neo4jPassword = neo4jPassword.read(sinkAnnotation),
+            topicQuerys = sinkAnnotation.topics.zip(sinkAnnotation.queries).toMap())
+    sink.register(kafkaConnectExternalUri.read(sinkAnnotation))
   }
 
   override fun afterEach(p0: ExtensionContext?) {
-    TODO("Not yet implemented")
+    sink.unregister()
   }
 }

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/sink/Neo4jSinkRegistration.kt

@@ -26,7 +26,7 @@ import org.neo4j.connectors.kafka.testing.RegistrationSupport.randomizedName
 class Neo4jSinkRegistration(
     topicQuerys: Map<String, String>,
     neo4jUri: String,
-    neo4jUsername: String,
+    neo4jUser: String,
     neo4jPassword: String,
     enableKeySchemas: Boolean = false,
     enableValueSchemas: Boolean = false,
@@ -59,7 +59,7 @@ class Neo4jSinkRegistration(
                         "errors.log.enable" to enableErrorLog,
                         "errors.log.include.messages" to includeMessagesInErrorLog,
                         "neo4j.server.uri" to neo4jUri,
-                        "neo4j.authentication.basic.username" to neo4jUsername,
+                        "neo4j.authentication.basic.username" to neo4jUser,
                         "neo4j.authentication.basic.password" to neo4jPassword,
                     ))
             .nestUnder("config", queries)
@@ -77,7 +77,7 @@ class Neo4jSinkRegistration(
     RegistrationSupport.unregisterConnector(URI("$connectBaseUri/connectors/$name/"))
   }
 
-  internal fun getPayload(): Map<String, Any> {
+  /*nocommit*/ fun getPayload(): Map<String, Any> {
     return payload
   }
 }

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/source/Neo4jSourceExtension.kt

@@ -59,28 +59,19 @@ internal class Neo4jSourceExtension(
   private lateinit var session: Session
 
   private val brokerExternalHost =
-      EnvBackedSetting<Neo4jSource>(
-          "brokerExternalHost", "BROKER_EXTERNAL_HOST", { it.brokerExternalHost })
+      EnvBackedSetting<Neo4jSource>("brokerExternalHost", { it.brokerExternalHost })
   private val schemaRegistryUri =
-      EnvBackedSetting<Neo4jSource>(
-          "schemaControlRegistryUri",
-          "SCHEMA_CONTROL_REGISTRY_URI",
-          { it.schemaControlRegistryUri })
+      EnvBackedSetting<Neo4jSource>("schemaControlRegistryUri", { it.schemaControlRegistryUri })
   private val schemaRegistryExternalUri =
       EnvBackedSetting<Neo4jSource>(
-          "schemaControlRegistryExternalUri",
-          "SCHEMA_CONTROL_REGISTRY_EXTERNAL_URI",
-          { it.schemaControlRegistryExternalUri })
+          "schemaControlRegistryExternalUri", { it.schemaControlRegistryExternalUri })
   private val kafkaConnectExternalUri =
-      EnvBackedSetting<Neo4jSource>(
-          "kafkaConnectExternalUri", "KAFKA_CONNECT_EXTERNAL_URI", { it.kafkaConnectExternalUri })
-  private val neo4jUri = EnvBackedSetting<Neo4jSource>("neo4jUri", "NEO4J_URI", { it.neo4jUri })
+      EnvBackedSetting<Neo4jSource>("kafkaConnectExternalUri", { it.kafkaConnectExternalUri })
+  private val neo4jUri = EnvBackedSetting<Neo4jSource>("neo4jUri", { it.neo4jUri })
   private val neo4jExternalUri =
-      EnvBackedSetting<Neo4jSource>(
-          "neo4jExternalUri", "NEO4J_EXTERNAL_URI", { it.neo4jExternalUri })
-  private val neo4jUser = EnvBackedSetting<Neo4jSource>("neo4jUser", "NEO4J_USER", { it.neo4jUser })
-  private val neo4jPassword =
-      EnvBackedSetting<Neo4jSource>("neo4jPassword", "NEO4J_PASSWORD", { it.neo4jPassword })
+      EnvBackedSetting<Neo4jSource>("neo4jExternalUri", { it.neo4jExternalUri })
+  private val neo4jUser = EnvBackedSetting<Neo4jSource>("neo4jUser", { it.neo4jUser })
+  private val neo4jPassword = EnvBackedSetting<Neo4jSource>("neo4jPassword", { it.neo4jPassword })
 
   private val envSettings =
       listOf(

testing/src/test/kotlin/org/neo4j/connectors/kafka/testing/KafkaConnectServer.kt

@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.connectors.kafka.testing
+
+import com.sun.net.httpserver.HttpExchange
+import com.sun.net.httpserver.HttpServer
+import java.net.InetSocketAddress
+
+class KafkaConnectServer : AutoCloseable {
+
+  companion object {
+    val CANDIDATE_ACCEPT_HEADERS = setOf("*/*", "application/*", "*/json", "application/json")
+  }
+
+  private var started = false
+  private val httpServer: HttpServer = HttpServer.create(InetSocketAddress(0), 0)
+
+  /**
+   * Starts the fake Kafka Connect server
+   *
+   * @param registrationHandler an optional handler to act on the ongoing HTTP exchange for
+   *   connector registration
+   * @param unregistrationHandler an optional handler to act on the ongoing HTTP exchange for
+   *   connector registration Note that both handlers should return true if
+   *   HttpExchange.sendResponseHeaders is called and false otherwise
+   */
+  fun start(
+      registrationHandler: (HttpExchange) -> Boolean = {
+        it.sendResponseHeaders(201, -1)
+        true
+      },
+      unregistrationHandler: (HttpExchange) -> Boolean = {
+        it.sendResponseHeaders(204, -1)
+        true
+      }
+  ) {
+
+    httpServer.createContext("/connectors") { exchange ->
+      val path = exchange.requestURI.path.replaceFirst("/connectors/?([^/]*)/?".toRegex(), "$1")
+
+      when {
+        path.isEmpty() -> handleRegistration(exchange, registrationHandler)
+        !path.contains("/") -> handleUnregistration(exchange, unregistrationHandler)
+        else -> exchange.sendResponseHeaders(404, 0)
+      }
+      exchange.close()
+    }
+    this.started = true
+    this.httpServer.start()
+  }
+
+  fun address(): String {
+    return "http://localhost:${httpServer.address.port}"
+  }
+
+  override fun close() {
+    if (this.started) {
+      this.started = false
+      httpServer.stop(0)
+    }
+  }
+
+  private fun handleRegistration(exchange: HttpExchange, handler: (HttpExchange) -> Boolean) {
+    if (!validateMethod(exchange, "POST") ||
+        !validateJsonIn(exchange) ||
+        !validateJsonOut(exchange)) {
+      return
+    }
+    if (!handler(exchange)) {
+      exchange.sendResponseHeaders(201, -1)
+    }
+  }
+
+  private fun handleUnregistration(exchange: HttpExchange, handler: (HttpExchange) -> Boolean) {
+    if (!validateMethod(exchange, "DELETE") || !validateJsonOut(exchange)) {
+      return
+    }
+    if (!handler(exchange)) {
+      exchange.sendResponseHeaders(204, -1)
+    }
+  }
+
+  private fun validateMethod(exchange: HttpExchange, method: String): Boolean {
+    if (exchange.requestMethod != method) {
+      exchange.sendResponseHeaders(405, -1)
+      return false
+    }
+    return true
+  }
+
+  private fun validateJsonIn(exchange: HttpExchange): Boolean {
+    val contentType = exchange.requestHeaders["Content-Type"]!!
+    if (contentType.size != 1) {
+      exchange.sendResponseHeaders(400, -1)
+      exchange.responseBody.bufferedWriter().write("unparseable Content-Type")
+      return false
+    }
+    if (contentType[0] != "application/json") {
+      exchange.sendResponseHeaders(415, -1)
+      return false
+    }
+    return true
+  }
+
+  private fun validateJsonOut(exchange: HttpExchange): Boolean {
+    val acceptedMedia = exchange.requestHeaders["Accept"]!!
+    if (acceptedMedia.isNotEmpty() &&
+        acceptedMedia.none { CANDIDATE_ACCEPT_HEADERS.contains(it) }) {
+      exchange.sendResponseHeaders(406, -1)
+      return false
+    }
+    return true
+  }
+}

testing/src/test/kotlin/org/neo4j/connectors/kafka/testing/WordSupportTest.kt

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.connectors.kafka.testing
+
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.Test
+
+class WordSupportTest {
+
+  @Test
+  fun pluralizes() {
+    assertEquals("flowers", WordSupport.pluralize(2, "flower", "flowers"))
+    assertEquals("alumnus", WordSupport.pluralize(1, "alumnus", "alumni"))
+    assertEquals("query", WordSupport.pluralize(0, "query", "queries"))
+    assertEquals("query", WordSupport.pluralize(0, "query", "queries"))
+    assertEquals("dollar", WordSupport.pluralize(-1, "dollar", "dollars"))
+    assertEquals("years", WordSupport.pluralize(-2, "year", "years"))
+  }
+
+  @Test
+  fun `converts camel case to upper snake case`() {
+    assertEquals("EASY", WordSupport.camelCaseToUpperSnakeCase("easy"))
+    assertEquals("THIS_IS_BANANAS", WordSupport.camelCaseToUpperSnakeCase("thisIsBananas"))
+  }
+}