Merge pull request #118 from aerospike/develop

v0.11.0

build.gradle.kts

@@ -9,7 +9,6 @@ plugins {
 }
 
 group = "com.aerospike"
-version = "0.10.0"
 
 repositories {
     mavenCentral()
@@ -31,30 +30,30 @@ tasks.withType<Jar> {
 }
 
 // Common dependency versions.
-extra["nettyVersion"] = "4.1.65.Final"
-extra["logbackVersion"] = "1.2.3"
-extra["jacksonVersion"] = "2.12.2"
+extra["nettyVersion"] = "4.1.72.Final"
+extra["logbackVersion"] = "1.2.10"
+extra["jacksonVersion"] = "2.13.1"
 
 dependencies {
-    implementation("com.aerospike:aerospike-client:5.1.8")
+    implementation("com.aerospike:aerospike-client:5.1.11")
     implementation("io.netty:netty-all:${project.extra["nettyVersion"]}")
     implementation("io.netty:netty-codec-redis:${project.extra["nettyVersion"]}")
     implementation("com.google.inject:guice:5.0.1")
-    implementation("io.github.microutils:kotlin-logging:2.0.6")
+    implementation("io.github.microutils:kotlin-logging:2.1.21")
     implementation("ch.qos.logback:logback-classic:${project.extra["logbackVersion"]}")
     implementation("ch.qos.logback:logback-core:${project.extra["logbackVersion"]}")
-    implementation("info.picocli:picocli:4.6.1")
-    implementation("commons-io:commons-io:2.8.0")
+    implementation("info.picocli:picocli:4.6.2")
+    implementation("commons-io:commons-io:2.11.0")
     implementation("com.fasterxml.jackson.core:jackson-core:${project.extra["jacksonVersion"]}")
     implementation("com.fasterxml.jackson.core:jackson-annotations:${project.extra["jacksonVersion"]}")
     implementation("com.fasterxml.jackson.core:jackson-databind:${project.extra["jacksonVersion"]}")
     implementation("com.fasterxml.jackson.module:jackson-module-kotlin:${project.extra["jacksonVersion"]}")
     implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:${project.extra["jacksonVersion"]}")
-    implementation("com.google.guava:guava:30.1.1-jre")
+    implementation("com.google.guava:guava:31.0.1-jre")
 
     testImplementation(kotlin("test-junit5"))
-    testImplementation("org.junit.jupiter:junit-jupiter-api:5.6.0")
-    testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.6.0")
+    testImplementation("org.junit.jupiter:junit-jupiter-api:5.8.2")
+    testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.8.2")
 }
 
 tasks.test {

gradle.properties

@@ -1 +1,2 @@
 kotlin.code.style=official
+version=0.11.0

pkg/install/etc/skyhook/skyhook.yml

@@ -1,14 +1,27 @@
 ---
-# The Skyhook server configuration
+# Skyhook server configuration
 
 hostList: localhost:3000
 namespace: test
 set: redis
 bin: b
 redisPort: 6379
 
+# Aerospike Java Client [com.aerospike.client.policy.ClientPolicy] configuration.
+#clientPolicy:
+#  user: admin
+#  password: pwd@1234
+#  clusterName: cluster1
+#  authMode: EXTERNAL_INSECURE
+#  timeout: 1500
+#  loginTimeout: 3000
+#  asyncMinConnsPerNode: 50
+#  asyncMaxConnsPerNode: 200
+#  failIfNotConnected: true
+#  useServicesAlternate: true
+
 # Bind on unix socket.
-# unixSocket: "/tmp/skyhook.sock"
+#unixSocket: "/tmp/skyhook.sock"
 
 workerThreads: 2
 bossThreads: 1

src/main/kotlin/com/aerospike/skyhook/SkyhookModule.kt

@@ -22,9 +22,9 @@ import io.netty.channel.socket.nio.NioServerSocketChannel
 import mu.KotlinLogging
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
-import javax.inject.Inject
 import javax.inject.Singleton
 
+@Suppress("unused")
 class SkyhookModule(
     private val config: ServerConfiguration
 ) : AbstractModule() {
@@ -93,21 +93,25 @@ class SkyhookModule(
 
     @Provides
     @Singleton
-    @Inject
     fun serverConfiguration(): ServerConfiguration {
         return config
     }
 
     @Provides
     @Singleton
-    @Inject
     fun clientPolicy(): ClientPolicy {
         val clientPolicy = ClientPolicy()
         clientPolicy.eventLoops = getClientEventLoops()
         config.clientPolicy.user?.let { clientPolicy.user = it }
         config.clientPolicy.password?.let { clientPolicy.password = it }
+        config.clientPolicy.clusterName?.let { clientPolicy.clusterName = it }
         config.clientPolicy.authMode?.let { clientPolicy.authMode = it }
         config.clientPolicy.timeout?.let { clientPolicy.timeout = it }
+        config.clientPolicy.loginTimeout?.let { clientPolicy.loginTimeout = it }
+        config.clientPolicy.asyncMinConnsPerNode?.let { clientPolicy.asyncMinConnsPerNode = it }
+        config.clientPolicy.asyncMaxConnsPerNode?.let { clientPolicy.asyncMaxConnsPerNode = it }
+        config.clientPolicy.failIfNotConnected?.let { clientPolicy.failIfNotConnected = it }
+        config.clientPolicy.useServicesAlternate?.let { clientPolicy.useServicesAlternate = it }
         return clientPolicy
     }
 
@@ -121,7 +125,6 @@ class SkyhookModule(
 
     @Provides
     @Singleton
-    @Inject
     fun executorService(): ExecutorService {
         return Executors.newFixedThreadPool(config.workerThreads)
     }

src/main/kotlin/com/aerospike/skyhook/command/RedisCommand.kt

@@ -252,7 +252,7 @@ enum class RedisCommand(
 
         fun getValue(stringValue: String): RedisCommand {
             return try {
-                valueOf(stringValue.toUpperCase(Locale.ENGLISH))
+                valueOf(stringValue.uppercase(Locale.ENGLISH))
             } catch (e: IllegalArgumentException) {
                 val msg = "ERR $stringValue unsupported command"
                 log.warn { msg }

src/main/kotlin/com/aerospike/skyhook/config/ClientPolicyConfig.kt

@@ -2,6 +2,9 @@ package com.aerospike.skyhook.config
 
 import com.aerospike.client.policy.AuthMode
 
+/**
+ * Aerospike Java Client [com.aerospike.client.policy.ClientPolicy] configuration properties.
+ */
 data class ClientPolicyConfig(
 
     /**
@@ -17,6 +20,13 @@ data class ClientPolicyConfig(
      */
     val password: String? = null,
 
+    /**
+     * Expected cluster name. If not null, server nodes must return this cluster name in order to join
+     * the client's view of the cluster. Should only be set when connecting to servers that support the
+     * "cluster-name" info command.
+     */
+    val clusterName: String? = null,
+
     /**
      * Authentication mode used when user/password is defined.
      */
@@ -26,5 +36,36 @@ data class ClientPolicyConfig(
      * Initial host connection timeout in milliseconds.
      * The timeout when opening a connection to the server host for the first time.
      */
-    val timeout: Int? = null
+    val timeout: Int? = null,
+
+    /**
+     * Login timeout in milliseconds. The timeout is used when user authentication is enabled
+     * and a node login is being performed.
+     */
+    val loginTimeout: Int? = null,
+
+    /**
+     * Minimum number of asynchronous connections allowed per server node.
+     * Preallocate min connections on client node creation. The client will periodically allocate new connections
+     * if count falls below min connections.
+     */
+    val asyncMinConnsPerNode: Int? = null,
+
+    /**
+     * Maximum number of asynchronous connections allowed per server node.
+     * Transactions will go through retry logic and potentially fail with "ResultCode.NO_MORE_CONNECTIONS"
+     * if the maximum number of connections would be exceeded.
+     */
+    val asyncMaxConnsPerNode: Int? = null,
+
+    /**
+     * Throw exception if all seed connections fail on cluster instantiation.
+     */
+    val failIfNotConnected: Boolean? = null,
+
+    /**
+     * Should use "services-alternate" instead of "services" in info request during cluster tending.
+     * "services-alternate" returns server configured external IP addresses that client uses to talk to nodes.
+     */
+    val useServicesAlternate: Boolean? = null,
 )

src/main/kotlin/com/aerospike/skyhook/config/ServerConfiguration.kt

@@ -1,5 +1,8 @@
 package com.aerospike.skyhook.config
 
+/**
+ * Skyhook server configuration properties.
+ */
 data class ServerConfiguration(
 
     /**
@@ -18,7 +21,7 @@ data class ServerConfiguration(
     val set: String? = "redis",
 
     /**
-     * Aerospike Client Policy configuration properties.
+     * Aerospike Java Client [com.aerospike.client.policy.ClientPolicy] configuration.
      */
     val clientPolicy: ClientPolicyConfig = ClientPolicyConfig(),
 

src/main/kotlin/com/aerospike/skyhook/handler/aerospike/DbsizeCommandHandler.kt

@@ -22,8 +22,7 @@ class DbsizeCommandHandler(
     private fun getTableRecordsNumber(ns: String, set: String?): Long {
         val allRecords = client.nodes
             .map { getSetInfo(ns, set, it) }
-            .map { it["objects"]!!.toInt() }
-            .sum()
+            .sumOf { it["objects"]!!.toInt() }
         val replicationFactor = getNamespaceInfo(ns, client.nodes[0])["effective_replication_factor"]!!.toInt()
         return floor(allRecords.toDouble() / replicationFactor).toLong()
     }

src/main/kotlin/com/aerospike/skyhook/handler/redis/CommandCommandHandler.kt

@@ -18,13 +18,13 @@ class CommandCommandHandler(
         if (cmd.argCount == 1) {
             RedisCommand.writeCommand(ctx)
         } else {
-            when (String(cmd.args[1]).toUpperCase(Locale.ENGLISH)) {
+            when (String(cmd.args[1]).uppercase(Locale.ENGLISH)) {
                 "COUNT" -> {
                     writeLong(RedisCommand.totalCommands)
                 }
                 "INFO" -> {
                     val commands = cmd.args.drop(2).map { String(it) }
-                        .map { it.toLowerCase(Locale.ENGLISH) }
+                        .map { it.lowercase(Locale.ENGLISH) }
                     RedisCommand.writeCommandInfo(ctx, commands)
                 }
                 else -> {

src/main/kotlin/com/aerospike/skyhook/listener/ValueType.kt

@@ -1,5 +1,7 @@
 package com.aerospike.skyhook.listener
 
+import java.util.*
+
 /**
  * Redis supported value types.
  */
@@ -12,6 +14,6 @@ enum class ValueType(val str: String) {
     STREAM("stream");
 
     companion object {
-        fun valueOf(ba: ByteArray) = valueOf(String(ba).toUpperCase())
+        fun valueOf(ba: ByteArray) = valueOf(String(ba).uppercase(Locale.ENGLISH))
     }
 }

src/main/kotlin/com/aerospike/skyhook/listener/key/GetCommandListener.kt

@@ -9,6 +9,7 @@ import com.aerospike.skyhook.command.RequestCommand
 import com.aerospike.skyhook.listener.BaseListener
 import com.aerospike.skyhook.listener.ValueType
 import io.netty.channel.ChannelHandlerContext
+import java.util.*
 
 open class GetCommandListener(
     ctx: ChannelHandlerContext
@@ -90,7 +91,7 @@ class GetexCommandListener(
 
         private fun setFlag(i: Int) {
             val flagStr = String(cmd.args[i])
-            when (flagStr.toUpperCase()) {
+            when (flagStr.uppercase(Locale.ENGLISH)) {
                 "EX" -> EX = String(cmd.args[i + 1]).toInt()
                 "PX" -> PX = String(cmd.args[i + 1]).toInt()
                 "EXAT" -> EXAT = String(cmd.args[i + 1]).toLong()

src/main/kotlin/com/aerospike/skyhook/listener/key/SetCommandListener.kt

@@ -75,7 +75,7 @@ class SetCommandListener(
 
         private fun setFlag(i: Int) {
             val flagStr = String(cmd.args[i])
-            when (flagStr.toUpperCase()) {
+            when (flagStr.uppercase(Locale.ENGLISH)) {
                 "EX" -> EX = String(cmd.args[i + 1]).toInt()
                 "PX" -> PX = String(cmd.args[i + 1]).toInt()
                 "EXAT" -> EXAT = String(cmd.args[i + 1]).toLong()

src/main/kotlin/com/aerospike/skyhook/listener/map/ZaddCommandListener.kt

@@ -14,6 +14,7 @@ import com.aerospike.skyhook.listener.BaseListener
 import com.aerospike.skyhook.listener.ValueType
 import com.aerospike.skyhook.util.Typed
 import io.netty.channel.ChannelHandlerContext
+import java.util.*
 
 class ZaddCommandListener(
     ctx: ChannelHandlerContext
@@ -47,7 +48,7 @@ class ZaddCommandListener(
         }
 
         private fun setFlag(flagStr: String): Boolean {
-            when (flagStr.toUpperCase()) {
+            when (flagStr.uppercase(Locale.ENGLISH)) {
                 "XX" -> XX = true
                 "NX" -> NX = true
                 "LT" -> LT = true

src/main/kotlin/com/aerospike/skyhook/listener/map/ZrangeCommandListener.kt

@@ -11,6 +11,7 @@ import com.aerospike.skyhook.command.RequestCommand
 import com.aerospike.skyhook.listener.BaseListener
 import com.aerospike.skyhook.util.Intervals
 import io.netty.channel.ChannelHandlerContext
+import java.util.*
 
 open class ZrangeCommandListener(
     ctx: ChannelHandlerContext
@@ -58,7 +59,7 @@ open class ZrangeCommandListener(
 
         private fun setFlag(i: Int): Int {
             val flagStr = String(cmd.args[i])
-            when (flagStr.toUpperCase()) {
+            when (flagStr.uppercase(Locale.ENGLISH)) {
                 "BYSCORE" -> BYSCORE = true
                 "BYLEX" -> BYLEX = true
                 "REV" -> REV = true

src/main/kotlin/com/aerospike/skyhook/listener/scan/ScanCommand.kt

@@ -3,6 +3,7 @@ package com.aerospike.skyhook.listener.scan
 import com.aerospike.skyhook.command.RequestCommand
 import com.aerospike.skyhook.listener.ValueType
 import com.aerospike.skyhook.util.RegexUtils
+import java.util.*
 
 class ScanCommand(val cmd: RequestCommand, flagIndex: Int) {
     var MATCH: String? = null
@@ -19,7 +20,7 @@ class ScanCommand(val cmd: RequestCommand, flagIndex: Int) {
 
     private fun setFlag(i: Int) {
         val flagStr = String(cmd.args[i])
-        when (flagStr.toUpperCase()) {
+        when (flagStr.uppercase(Locale.ENGLISH)) {
             "MATCH" -> MATCH = RegexUtils.format(String(cmd.args[i + 1]))
             "COUNT" -> COUNT = String(cmd.args[i + 1]).toLong()
             "TYPE" -> TYPE = ValueType.valueOf(cmd.args[i + 1])

src/main/kotlin/com/aerospike/skyhook/util/InfoUtils.kt

@@ -9,15 +9,14 @@ object InfoUtils {
     fun getNamespaceInfo(ns: String, node: Node): Map<String, String> {
         val schemaInfo = Info.request(null, node, "namespace/$ns")
         return schemaInfo.split(";")
-            .map { it.split("=".toRegex(), 2).toPair() }
-            .toMap()
+            .associate { it.split("=".toRegex(), 2).toPair() }
     }
 
     fun getSetInfo(ns: String, set: String?, node: Node): Map<String, String> {
         val sets = Info.request(null, node, "sets")
         val tableInfo = sets.split(";")
             .filter { it.startsWith("ns=$ns:set=$set") }[0]
         return Pattern.compile("\\s*:\\s*").split(tableInfo).toList()
-            .filterNotNull().map { it.split("=".toRegex(), 2).toPair() }.toMap()
+            .filterNotNull().associate { it.split("=".toRegex(), 2).toPair() }
     }
 }

src/main/kotlin/com/aerospike/skyhook/util/Intervals.kt

@@ -1,5 +1,7 @@
 package com.aerospike.skyhook.util
 
+import java.util.*
+
 object Intervals {
 
     private const val infHighest = "+inf"
@@ -20,7 +22,7 @@ object Intervals {
     }
 
     private fun score(interval: String, includeShift: Int, excludeShift: Int): Int {
-        return when (interval.toLowerCase()) {
+        return when (interval.lowercase(Locale.ENGLISH)) {
             infHighest -> Int.MAX_VALUE
             infLowest -> Int.MIN_VALUE
             else -> {
@@ -42,7 +44,7 @@ object Intervals {
     }
 
     private fun lex(interval: String, from: Boolean): String {
-        return when (interval.toLowerCase()) {
+        return when (interval.lowercase(Locale.ENGLISH)) {
             lexHighest -> String(byteArrayOf(127))
             lexLowest -> String(byteArrayOf(0))
             else -> {

src/main/kotlin/com/aerospike/skyhook/util/SystemUtils.kt

@@ -1,13 +1,15 @@
 package com.aerospike.skyhook.util
 
+import java.util.*
+
 object SystemUtils {
 
     enum class OS {
         LINUX, MAC, WINDOWS, OTHER
     }
 
     val os: OS by lazy {
-        val os = System.getProperty("os.name").toLowerCase()
+        val os = System.getProperty("os.name").lowercase(Locale.ENGLISH)
         when {
             os.contains("nux") -> {
                 OS.LINUX