Bump dependency versions to the latest and the JDK version to 17, and use `PgConnection` instead of `PgPool` in the benchmark portion "vertx-web-scala" (#7934)

* Upgrade the dependency versions to the latest supported by Scala (Scala 2.12.17, SBT 1.8.2, Vert.x 3.9.15, Vert.x Scala libraries 3.9.1) and the JDK version to 17 in the vertx-web-scala portion

* Use `PgConnection` instead of `PgPool` in the "vertx-web-scala" portion

The PG pipelining limit is set to 100000 to be consistent with the other benchmarks.

frameworks/Scala/vertx-web-scala/.gitignore

@@ -183,3 +183,5 @@ nbdist/
 !.vscode/tasks.json
 !.vscode/launch.json
 !.vscode/extensions.json
+
+.bsp

frameworks/Scala/vertx-web-scala/README.md

@@ -13,8 +13,8 @@ This is the Vert.x Web Scala portion of a [benchmarking test suite](../) compari
 
 ## Versions
 
-* [Java OpenJDK 11](https://openjdk.java.net/)
-* [Vert.x 3.7](https://vertx.io/)
+* [Java OpenJDK 17](https://openjdk.java.net/)
+* [Vert.x 3.9](https://vertx.io/)
 
 ## Test URLs
 

frameworks/Scala/vertx-web-scala/build.sbt

@@ -2,22 +2,26 @@ name := "vertx-web-scala"
 
 version := "1"
 
-scalaVersion := "2.12.8"
+scalaVersion := "2.12.17"
 
 lazy val root = (project in file(".")).enablePlugins(SbtTwirl)
 
-libraryDependencies += "io.vertx" %% "vertx-lang-scala" % "3.7.1"
-libraryDependencies += "io.vertx" %% "vertx-web-scala" % "3.7.1"
-libraryDependencies += "io.vertx" % "vertx-codegen" % "3.7.1"
-libraryDependencies += "io.netty" % "netty-transport-native-kqueue" % "4.1.36.Final" classifier "osx-x86_64"
-libraryDependencies += "io.netty" % "netty-transport-native-epoll" % "4.1.36.Final" classifier "linux-x86_64"
-libraryDependencies += "io.reactiverse" % "reactive-pg-client" % "0.11.3"
-libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.2.3"
-libraryDependencies += "com.typesafe.scala-logging" %% "scala-logging" % "3.9.2"
+val vertxScalaVersion = "3.9.1"
+val vertxVersion = "3.9.15"
+val nettyVersion = "4.1.89.Final"
 
-mainClass in Compile := Some("vertx.App")
+libraryDependencies += "io.vertx" %% "vertx-lang-scala" % vertxScalaVersion
+libraryDependencies += "io.vertx" %% "vertx-web-scala" % vertxScalaVersion
+libraryDependencies += "io.vertx" % "vertx-codegen" % vertxVersion
+libraryDependencies += "io.vertx" % "vertx-pg-client" % vertxVersion
+libraryDependencies += "io.netty" % "netty-transport-native-kqueue" % nettyVersion classifier "osx-x86_64"
+libraryDependencies += "io.netty" % "netty-transport-native-epoll" % nettyVersion classifier "linux-x86_64"
+libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.4.5"
+libraryDependencies += "com.typesafe.scala-logging" %% "scala-logging" % "3.9.5"
 
-assemblyMergeStrategy in assembly := {
+Compile / mainClass := Some("vertx.App")
+
+assembly / assemblyMergeStrategy := {
   case PathList("META-INF", "MANIFEST.MF") => MergeStrategy.discard
-  case _ => MergeStrategy.first
-}
+  case _                                   => MergeStrategy.first
+}

frameworks/Scala/vertx-web-scala/project/build.properties

@@ -1 +1 @@
-sbt.version=1.2.8
+sbt.version=1.8.2

frameworks/Scala/vertx-web-scala/project/plugins.sbt

@@ -1,3 +1,3 @@
-addSbtPlugin("com.typesafe.sbt" % "sbt-twirl" % "1.4.1")
-addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.0.1")
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.9")
+addSbtPlugin("com.typesafe.play" % "sbt-twirl" % "1.5.2")
+addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.6")
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.1")

frameworks/Scala/vertx-web-scala/src/main/scala/vertx/App.scala

@@ -7,18 +7,20 @@ import java.time.format.DateTimeFormatter
 import java.util.concurrent.ThreadLocalRandom
 
 import com.typesafe.scalalogging.Logger
-import io.reactiverse.pgclient._
 import io.vertx.core.buffer.Buffer
 import io.vertx.core.http.HttpHeaders
 import io.vertx.core.json.{JsonArray, JsonObject}
-import io.vertx.core.{AsyncResult, VertxOptions => JVertxOptions}
+import io.vertx.core.{AsyncResult, Handler, VertxOptions => JVertxOptions}
 import io.vertx.lang.scala.{ScalaVerticle, VertxExecutionContext}
+import io.vertx.pgclient._
 import io.vertx.scala.core.http.{HttpServer, HttpServerRequest, HttpServerResponse}
 import io.vertx.scala.core.{VertxOptions, _}
 import io.vertx.scala.ext.web.Router
+import io.vertx.sqlclient._
 import vertx.model.{Fortune, Message, World}
 
 import scala.collection.JavaConverters._
+import scala.concurrent.{Future, Promise}
 import scala.util.{Failure, Sorting, Success, Try}
 
 case class Header(name: CharSequence, value: String)
@@ -26,7 +28,7 @@ case class Header(name: CharSequence, value: String)
 class App extends ScalaVerticle {
 
   private val HELLO_WORLD = "Hello, world!"
-  private val HELLO_WORLD_BUFFER = Buffer.factory.directBuffer(HELLO_WORLD, "UTF-8")
+  private val HELLO_WORLD_BUFFER = Buffer.buffer(HELLO_WORLD, "UTF-8")
   private val SERVER = "vert.x"
 
   private val contentTypeJson = Header(HttpHeaders.CONTENT_TYPE, "application/json")
@@ -36,26 +38,33 @@ class App extends ScalaVerticle {
   private var dateString: String = ""
 
   private var server: HttpServer = _
-  private var client: PgClient = _
-  private var pool: PgPool = _
+  private var client: PgConnection = _
 
   private def refreshDateHeader(): Unit = dateString = App.createDateHeader()
 
-  override def start(): Unit = {
+  override def startFuture(): Future[_] = {
     refreshDateHeader()
     vertx.setPeriodic(1000, (_: Long) => refreshDateHeader())
 
-    val options = new PgPoolOptions()
+    val pgConnectOptions = new PgConnectOptions()
       .setDatabase(config.getString("database"))
       .setHost(config.getString("host"))
       .setPort(config.getInteger("port", 5432))
       .setUser(config.getString("username"))
       .setPassword(config.getString("password"))
       .setCachePreparedStatements(true)
+      .setPipeliningLimit(100000)
 
     val jVertx = vertx.asJava.asInstanceOf[io.vertx.core.Vertx]
-    client = PgClient.pool(jVertx, new PgPoolOptions(options).setMaxSize(1))
-    pool = PgClient.pool(jVertx, new PgPoolOptions(options).setMaxSize(4))
+    val pgConnectionPromise = Promise[Unit]
+    PgConnection.connect(
+      jVertx,
+      pgConnectOptions,
+      (ar => {
+        client = ar.result()
+        pgConnectionPromise.success()
+      }): Handler[AsyncResult[PgConnection]]
+    )
 
     val router = Router.router(vertx)
     router.get("/plaintext").handler(context => handlePlainText(context.request()))
@@ -67,7 +76,12 @@ class App extends ScalaVerticle {
 
     val port = 8080
     server = vertx.createHttpServer()
-    server.requestHandler(router.accept).listen(port)
+    val httpServerPromise = Promise[Unit]
+    server
+      .requestHandler(router.accept)
+      .listen(port, (_ => httpServerPromise.success()): Handler[AsyncResult[HttpServer]])
+
+    pgConnectionPromise.future.flatMap(_ => httpServerPromise.future)
   }
 
   override def stop(): Unit = Option(server).foreach(_.close())
@@ -87,25 +101,26 @@ class App extends ScalaVerticle {
       .end(Message("Hello, World!").toBuffer)
 
   private def handleDb(request: HttpServerRequest): Unit =
-    client.preparedQuery(
-      "SELECT id, randomnumber from WORLD where id=$1",
-      Tuple.of(App.randomWorld(), Nil: _*),
-      (ar: AsyncResult[PgRowSet]) => {
-        if (ar.succeeded) {
-          val resultSet = ar.result.iterator
-          if (!resultSet.hasNext) {
-            request.response.setStatusCode(404).end()
+    client
+      .preparedQuery("SELECT id, randomnumber from WORLD where id=$1")
+      .execute(
+        Tuple.of(App.randomWorld(), Nil: _*),
+        (ar: AsyncResult[RowSet[Row]]) => {
+          if (ar.succeeded) {
+            val resultSet = ar.result.iterator
+            if (!resultSet.hasNext) {
+              request.response.setStatusCode(404).end()
+            } else {
+              val row = resultSet.next
+              responseWithHeaders(request.response, contentTypeJson)
+                .end(World(row.getInteger(0), row.getInteger(1)).encode())
+            }
           } else {
-            val row = resultSet.next
-            responseWithHeaders(request.response, contentTypeJson)
-              .end(World(row.getInteger(0), row.getInteger(1)).encode())
+            App.logger.error("Failed to handle request", ar.cause())
+            request.response.setStatusCode(500).end(ar.cause.getMessage)
           }
-        } else {
-          App.logger.error("Failed to handle request", ar.cause())
-          request.response.setStatusCode(500).end(ar.cause.getMessage)
         }
-      }
-    )
+      )
 
   private def handleQueries(request: HttpServerRequest): Unit = {
     val queries = App.getQueries(request)
@@ -113,26 +128,27 @@ class App extends ScalaVerticle {
     var i = 0
     var failed = false
     while (i < queries) {
-      client.preparedQuery(
-        "SELECT id, randomnumber from WORLD where id=$1",
-        Tuple.of(App.randomWorld(), Nil: _*),
-        (ar: AsyncResult[PgRowSet]) => {
-          if (!failed) {
-            if (ar.failed) {
-              failed = true
-              request.response.setStatusCode(500).end(ar.cause.getMessage)
-              return
-            }
-            // we need a final reference
-            val row = ar.result.iterator.next
+      client
+        .preparedQuery("SELECT id, randomnumber from WORLD where id=$1")
+        .execute(
+          Tuple.of(App.randomWorld(), Nil: _*),
+          (ar: AsyncResult[RowSet[Row]]) => {
+            if (!failed) {
+              if (ar.failed) {
+                failed = true
+                request.response.setStatusCode(500).end(ar.cause.getMessage)
+                return
+              }
+              // we need a final reference
+              val row = ar.result.iterator.next
 
-            worlds.add(World(row.getInteger(0), row.getInteger(1)))
-            if (worlds.size == queries)
-              responseWithHeaders(request.response, contentTypeJson)
-                .end(worlds.encode)
+              worlds.add(World(row.getInteger(0), row.getInteger(1)))
+              if (worlds.size == queries)
+                responseWithHeaders(request.response, contentTypeJson)
+                  .end(worlds.encode)
+            }
           }
-        }
-      )
+        )
 
       i += 1
     }
@@ -144,24 +160,24 @@ class App extends ScalaVerticle {
       request.response.setStatusCode(500).end(err.getMessage)
     }
 
-    def handleUpdates(conn: PgConnection, worlds: Array[World]): Unit = {
+    def handleUpdates(conn: SqlConnection, worlds: Array[World]): Unit = {
       Sorting.quickSort(worlds)
 
       val batch = worlds.map(world => Tuple.of(world.randomNumber, world.id)).toList.asJava
-      conn.preparedBatch(
-        "UPDATE world SET randomnumber=$1 WHERE id=$2",
-        batch,
-        (ar: AsyncResult[PgRowSet]) => {
-          conn.close()
-          if (ar.failed) {
-            sendError(ar.cause)
-            return
-          }
+      conn
+        .preparedQuery("UPDATE world SET randomnumber=$1 WHERE id=$2")
+        .executeBatch(
+          batch,
+          (ar: AsyncResult[RowSet[Row]]) => {
+            if (ar.failed) {
+              sendError(ar.cause)
+              return
+            }
 
-          responseWithHeaders(request.response, contentTypeJson)
-            .end(new JsonArray(worlds.toList.asJava).toBuffer)
-        }
-      )
+            responseWithHeaders(request.response, contentTypeJson)
+              .end(new JsonArray(worlds.toList.asJava).toBuffer)
+          }
+        )
     }
 
     val queries = App.getQueries(request)
@@ -169,65 +185,57 @@ class App extends ScalaVerticle {
     var failed = false
     var queryCount = 0
 
-    pool.getConnection((ar1: AsyncResult[PgConnection]) => {
-      if (ar1.failed) {
-        failed = true
-        sendError(ar1.cause)
-        return
-      }
-      val conn = ar1.result
-      var i = 0
-      while (i < worlds.length) {
-        val id = App.randomWorld()
-        val index = i
-        conn.preparedQuery(
-          "SELECT id, randomnumber from WORLD where id=$1",
+    var i = 0
+    while (i < worlds.length) {
+      val id = App.randomWorld()
+      val index = i
+      client
+        .preparedQuery("SELECT id, randomnumber from WORLD where id=$1")
+        .execute(
           Tuple.of(id, Nil: _*),
-          (ar2: AsyncResult[PgRowSet]) => {
+          (ar2: AsyncResult[RowSet[Row]]) => {
             if (!failed) {
               if (ar2.failed) {
-                conn.close()
                 failed = true
                 sendError(ar2.cause)
                 return
               }
               worlds(index) = World(ar2.result.iterator.next.getInteger(0), App.randomWorld())
               queryCount += 1
-              if (queryCount == worlds.length) handleUpdates(conn, worlds)
+              if (queryCount == worlds.length) handleUpdates(client, worlds)
             }
           }
         )
 
-        i += 1
-      }
-
-    })
+      i += 1
+    }
   }
 
   private def handleFortunes(request: HttpServerRequest): Unit =
-    client.preparedQuery(
-      "SELECT id, message from FORTUNE",
-      (ar: AsyncResult[PgRowSet]) => {
-        val response = request.response
-        if (ar.succeeded) {
-          val resultSet = ar.result.iterator
-          if (!resultSet.hasNext) {
-            response.setStatusCode(404).end("No results")
-            return
+    client
+      .preparedQuery("SELECT id, message from FORTUNE")
+      .execute(
+        (ar: AsyncResult[RowSet[Row]]) => {
+          val response = request.response
+          if (ar.succeeded) {
+            val resultSet = ar.result.iterator
+            if (!resultSet.hasNext) {
+              response.setStatusCode(404).end("No results")
+              return
+            }
+            val fortunes = (resultSet.asScala
+              .map(row => Fortune(row.getInteger(0), row.getString(1))) ++
+              Seq(Fortune(0, "Additional fortune added at request time."))).toArray
+            Sorting.quickSort(fortunes)
+            responseWithHeaders(request.response, contentTypeHtml)
+              .end(html.fortune(fortunes).body)
+          } else {
+            val err = ar.cause
+            App.logger.error("", err)
+            response.setStatusCode(500).end(err.getMessage)
           }
-          val fortunes = (resultSet.asScala
-            .map(row => Fortune(row.getInteger(0), row.getString(1))) ++
-            Seq(Fortune(0, "Additional fortune added at request time."))).toArray
-          Sorting.quickSort(fortunes)
-          responseWithHeaders(request.response, contentTypeHtml)
-            .end(html.fortune(fortunes).body)
-        } else {
-          val err = ar.cause
-          App.logger.error("", err)
-          response.setStatusCode(500).end(err.getMessage)
         }
-      }
-    )
+      )
 
 }
 

frameworks/Scala/vertx-web-scala/vertx-web-scala.dockerfile

@@ -1,13 +1,6 @@
-FROM openjdk:11-jdk
+FROM sbtscala/scala-sbt:eclipse-temurin-17.0.5_8_1.8.2_2.12.17
 
-ARG SBT_VERSION=1.2.8
-
-# Install sbt
-RUN echo "deb https://repo.scala-sbt.org/scalasbt/debian all main" | tee /etc/apt/sources.list.d/sbt.list
-RUN echo "deb https://repo.scala-sbt.org/scalasbt/debian /" | tee /etc/apt/sources.list.d/sbt_old.list
-RUN curl -sL "https://keyserver.ubuntu.com/pks/lookup?op=get&search=0x2EE0EA64E40A89B84B2DF73499E82A75642AC823" | apt-key add
-RUN apt-get update -yqq
-RUN apt-get install -yqq sbt
+ARG SBT_VERSION=1.8.2
 
 WORKDIR /vertx
 COPY src src
@@ -33,4 +26,4 @@ CMD export DBIP=`getent hosts tfb-database | awk '{ print $1 }'` && \
       -Dvertx.disableHttpHeadersValidation=true \
       -jar \
       target/scala-2.12/vertx-web-scala-assembly-1.jar \
-      src/main/conf/config.json
+      src/main/conf/config.json