feat: add neo4j version parser

common/src/main/kotlin/org/neo4j/connectors/kafka/configuration/Neo4jVersion.kt

@@ -0,0 +1,107 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [https://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.configuration
+
+import java.util.*
+import kotlin.math.sign
+
+class Neo4jVersion(
+    private val major: Int,
+    private val minor: Int,
+    private val patch: Int = Int.MAX_VALUE
+) : Comparable<Neo4jVersion> {
+
+  override fun compareTo(other: Neo4jVersion): Int {
+    if (major != other.major) {
+      return signum(major - other.major)
+    }
+    if (minor != other.minor) {
+      return signum(minor - other.minor)
+    }
+    return signum(patch - other.patch)
+  }
+
+  override fun equals(other: Any?): Boolean {
+    if (other !is Neo4jVersion) {
+      return false
+    }
+    return major == other.major && minor == other.minor && patch == other.patch
+  }
+
+  override fun hashCode(): Int {
+    return Objects.hash(major, minor, patch)
+  }
+
+  override fun toString(): String {
+    if (patch == Int.MAX_VALUE) {
+      return String.format("%d.%d", major, minor)
+    }
+    return String.format("%d.%d.%d", major, minor, patch)
+  }
+
+  companion object {
+    val v5 = Neo4jVersion(5, 0, 0)
+    val v4_4 = Neo4jVersion(4, 4, 0)
+
+    fun of(version: String): Neo4jVersion {
+      var major = -1
+      var minor = -1
+      var patch = -1
+      var buffer = ""
+      for (c in version.toCharArray()) {
+        if (c != '.') {
+          buffer += c
+          continue
+        }
+        if (major == -1) {
+          major = buffer.toInt(10)
+        } else if (minor == -1) {
+          minor = parseMinor(buffer)
+        } else {
+          throw invalidVersion(version)
+        }
+        buffer = ""
+      }
+      if (buffer.isNotEmpty()) {
+        if (minor == -1) {
+          minor = parseMinor(buffer)
+        } else {
+          patch = buffer.toInt(10)
+        }
+      }
+      if (major == -1 || minor == -1) {
+        throw invalidVersion(version)
+      }
+      if (patch == -1) {
+        return Neo4jVersion(major, minor)
+      }
+      return Neo4jVersion(major, minor, patch)
+    }
+
+    private fun parseMinor(buffer: String): Int {
+      return buffer.replace("-aura", "").toInt(10)
+    }
+
+    private fun signum(result: Int): Int {
+      return sign(result.toDouble()).toInt()
+    }
+
+    private fun invalidVersion(version: String): IllegalArgumentException {
+      return IllegalArgumentException(String.format("Invalid Neo4j version: %s", version))
+    }
+  }
+}

common/src/test/kotlin/org/neo4j/connectors/kafka/configuration/Neo4jVersionTest.kt

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [https://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.configuration
+
+import kotlin.test.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+
+class Neo4jVersionTest {
+
+  @Test
+  fun `should parse version string`() {
+    assertEquals(Neo4jVersion.of("4.4"), Neo4jVersion(4, 4))
+    assertEquals(Neo4jVersion.of("4.4-aura"), Neo4jVersion(4, 4))
+    assertEquals(Neo4jVersion.of("4.4.13"), Neo4jVersion(4, 4, 13))
+    assertEquals(Neo4jVersion.of("2025.01"), Neo4jVersion(2025, 1))
+    assertEquals(Neo4jVersion.of("2025.01.0"), Neo4jVersion(2025, 1, 0))
+    assertEquals(Neo4jVersion.of("2025.01-aura"), Neo4jVersion(2025, 1))
+  }
+
+  @Test
+  fun `should fail invalid version format`() {
+    assertThrows<IllegalArgumentException> { Neo4jVersion.of("") }
+    assertThrows<IllegalArgumentException> { Neo4jVersion.of("5") }
+    assertThrows<IllegalArgumentException> { Neo4jVersion.of("5.5.3.1") }
+    assertThrows<NumberFormatException> { Neo4jVersion.of("..") }
+    assertThrows<NumberFormatException> { Neo4jVersion.of("5..3") }
+    assertThrows<NumberFormatException> { Neo4jVersion.of(".2025.6") }
+  }
+}

sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/SinkConfiguration.kt

@@ -25,6 +25,7 @@ import org.apache.kafka.connect.sink.SinkTask
 import org.neo4j.connectors.kafka.configuration.ConnectorType
 import org.neo4j.connectors.kafka.configuration.Groups
 import org.neo4j.connectors.kafka.configuration.Neo4jConfiguration
+import org.neo4j.connectors.kafka.configuration.Neo4jVersion
 import org.neo4j.connectors.kafka.configuration.helpers.ConfigKeyBuilder
 import org.neo4j.connectors.kafka.configuration.helpers.Recommenders
 import org.neo4j.connectors.kafka.configuration.helpers.SIMPLE_DURATION_PATTERN
@@ -79,7 +80,7 @@ class SinkConfiguration : Neo4jConfiguration {
   val patternBindValueAs
     get(): String = getString(PATTERN_BIND_VALUE_AS)
 
-  val dialect: Dialect by lazy {
+  private val dialect: Dialect by lazy {
     driver.session(sessionConfig()).use {
       val name = Cypher.name("name")
       val versions = Cypher.name("versions")
@@ -90,14 +91,11 @@ class SinkConfiguration : Neo4jConfiguration {
               .returning(Cypher.valueAt(versions, 0))
               .build()
 
-      val version = it.run(stmt.cypher, stmt.parameters).single().get(0).asString()
-      if (version.startsWith("5")) {
-        return@lazy Dialect.NEO4J_5
-      } else if (version.startsWith("4")) {
+      val version = Neo4jVersion.of(it.run(stmt.cypher, stmt.parameters).single().get(0).asString())
+      if (version < Neo4jVersion.v5) {
         return@lazy Dialect.DEFAULT
       }
-
-      throw ConfigException("unsupported Neo4j version: $version")
+      return@lazy Dialect.NEO4J_5
     }
   }