diff --git a/.gitignore b/.gitignore
index f266b41f..b83da02b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -20,6 +20,7 @@ postgresql-async/out/*
 mysql-async/target/*
 pool-async/target/*
 postgis-jasync/target/*
+r2dbc-mysql/target/*
 .rvmrc
 .ruby-version
 .ruby-gemset
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt
index 43b04677..4688f847 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLConnectionHandler.kt
@@ -4,6 +4,7 @@ import com.github.jasync.sql.db.Configuration
 import com.github.jasync.sql.db.exceptions.DatabaseException
 import com.github.jasync.sql.db.general.MutableResultSet
 import com.github.jasync.sql.db.mysql.binary.BinaryRowDecoder
+import com.github.jasync.sql.db.mysql.encoder.auth.AuthenticationMethod
 import com.github.jasync.sql.db.mysql.message.client.AuthenticationSwitchResponse
 import com.github.jasync.sql.db.mysql.message.client.CapabilityRequestMessage
 import com.github.jasync.sql.db.mysql.message.client.CloseStatementMessage
@@ -13,6 +14,7 @@ import com.github.jasync.sql.db.mysql.message.client.PreparedStatementPrepareMes
 import com.github.jasync.sql.db.mysql.message.client.QueryMessage
 import com.github.jasync.sql.db.mysql.message.client.QuitMessage
 import com.github.jasync.sql.db.mysql.message.client.SendLongDataMessage
+import com.github.jasync.sql.db.mysql.message.server.AuthMoreDataMessage
 import com.github.jasync.sql.db.mysql.message.server.AuthenticationSwitchRequest
 import com.github.jasync.sql.db.mysql.message.server.BinaryRowMessage
 import com.github.jasync.sql.db.mysql.message.server.ColumnDefinitionMessage
@@ -23,6 +25,7 @@ import com.github.jasync.sql.db.mysql.message.server.OkMessage
 import com.github.jasync.sql.db.mysql.message.server.PreparedStatementPrepareResponse
 import com.github.jasync.sql.db.mysql.message.server.ResultSetRowMessage
 import com.github.jasync.sql.db.mysql.message.server.ServerMessage
+import com.github.jasync.sql.db.mysql.util.CapabilityFlag
 import com.github.jasync.sql.db.mysql.util.CharsetMapper
 import com.github.jasync.sql.db.util.ExecutorServiceUtils
 import com.github.jasync.sql.db.util.FP
@@ -72,6 +75,7 @@ class MySQLConnectionHandler(
     private val parsedStatements = HashMap<String, PreparedStatementHolder>()
     private val binaryRowDecoder = BinaryRowDecoder()
 
+    private var sslEstablished: Boolean = false
     private var currentPreparedStatementHolder: PreparedStatementHolder? = null
     private var currentPreparedStatement: PreparedStatement? = null
     private var currentQuery: MutableResultSet<ColumnDefinitionMessage>? = null
@@ -127,6 +131,20 @@ class MySQLConnectionHandler(
                     ServerMessage.EOF -> {
                         this.handleEOF(message)
                     }
+                    ServerMessage.AuthMoreData -> {
+                        val m = message as AuthMoreDataMessage
+
+                        if (!m.isSuccess()) {
+                            if (!sslEstablished) {
+                                throw IllegalStateException(
+                                    "Full authentication mode for ${AuthenticationMethod.CachingSha2} requires SSL"
+                                )
+                            }
+
+                            val request = AuthenticationSwitchRequest(AuthenticationMethod.CachingSha2, null)
+                            handlerDelegate.switchAuthentication(request)
+                        }
+                    }
                     ServerMessage.ColumnDefinition -> {
                         val m = message as ColumnDefinitionMessage
 
@@ -278,6 +296,7 @@ class MySQLConnectionHandler(
     fun write(message: CapabilityRequestMessage): ChannelFuture = writeAndHandleError(message)
 
     fun write(message: HandshakeResponseMessage): ChannelFuture {
+        sslEstablished = message.header.flags.contains(CapabilityFlag.CLIENT_SSL)
         decoder.hasDoneHandshake = true
         return writeAndHandleError(message)
     }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLFrameDecoder.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLFrameDecoder.kt
index c0376632..1c92570f 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLFrameDecoder.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/codec/MySQLFrameDecoder.kt
@@ -3,6 +3,7 @@ package com.github.jasync.sql.db.mysql.codec
 import com.github.jasync.sql.db.exceptions.BufferNotFullyConsumedException
 import com.github.jasync.sql.db.exceptions.NegativeMessageSizeException
 import com.github.jasync.sql.db.exceptions.ParserNotAvailableException
+import com.github.jasync.sql.db.mysql.decoder.AuthMoreDataDecoder
 import com.github.jasync.sql.db.mysql.decoder.AuthenticationSwitchRequestDecoder
 import com.github.jasync.sql.db.mysql.decoder.ColumnDefinitionDecoder
 import com.github.jasync.sql.db.mysql.decoder.ColumnProcessingFinishedDecoder
@@ -39,6 +40,7 @@ class MySQLFrameDecoder(val charset: Charset, private val connectionId: String)
     private val handshakeDecoder = HandshakeV10Decoder()
     private val errorDecoder = ErrorDecoder(charset)
     private val okDecoder = OkDecoder(charset)
+    private val authMoreDataDecoder = AuthMoreDataDecoder()
     private val columnDecoder = ColumnDefinitionDecoder(charset, DecoderRegistry(charset))
     private val authenticationSwitchRequestDecoder = AuthenticationSwitchRequestDecoder(charset)
     private val rowDecoder = ResultSetRowDecoder()
@@ -89,7 +91,7 @@ class MySQLFrameDecoder(val charset: Charset, private val connectionId: String)
             logger.trace {
                 "[connectionId:$connectionId] - Reading message type $messageType - " +
                     "(count=$messagesCount,hasDoneHandshake=$hasDoneHandshake,size=$size,isInQuery=$isInQuery,processingColumns=$processingColumns,processingParams=$processingParams,processedColumns=$processedColumns,processedParams=$processedParams)" +
-                    "\n${BufferDumper.dumpAsHex(slice)}}"
+                    "\n${BufferDumper.dumpAsHex(slice)}"
             }
 
             slice.markReaderIndex()
@@ -161,6 +163,13 @@ class MySQLFrameDecoder(val charset: Charset, private val connectionId: String)
                     }
                 }
             }
+            ServerMessage.AuthMoreData -> {
+                if (!isInQuery) {
+                    this.authMoreDataDecoder
+                } else {
+                    null
+                }
+            }
             else -> {
                 if (this.isInQuery) {
                     null
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/decoder/AuthMoreDataDecoder.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/decoder/AuthMoreDataDecoder.kt
new file mode 100644
index 00000000..32c069e7
--- /dev/null
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/decoder/AuthMoreDataDecoder.kt
@@ -0,0 +1,13 @@
+package com.github.jasync.sql.db.mysql.decoder
+
+import com.github.jasync.sql.db.mysql.message.server.AuthMoreDataMessage
+import com.github.jasync.sql.db.mysql.message.server.ServerMessage
+import io.netty.buffer.ByteBuf
+
+class AuthMoreDataDecoder : MessageDecoder {
+    override fun decode(buffer: ByteBuf): ServerMessage {
+        return AuthMoreDataMessage(
+            data = buffer.readByte(),
+        )
+    }
+}
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/decoder/AuthenticationSwitchRequestDecoder.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/decoder/AuthenticationSwitchRequestDecoder.kt
index 635b85c2..97c0840e 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/decoder/AuthenticationSwitchRequestDecoder.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/decoder/AuthenticationSwitchRequestDecoder.kt
@@ -12,11 +12,12 @@ class AuthenticationSwitchRequestDecoder(val charset: Charset) : MessageDecoder
         val method = buffer.readCString(charset)
         val bytes: Int = buffer.readableBytes()
         val terminal = 0.toByte()
-        val salt = if (bytes > 0 && buffer.getByte(buffer.writerIndex() - 1) == terminal) ByteBufUtil.getBytes(
+        val seed = if (bytes > 0 && buffer.getByte(buffer.writerIndex() - 1) == terminal) ByteBufUtil.getBytes(
             buffer,
             buffer.readerIndex(),
             bytes - 1
         ) else ByteBufUtil.getBytes(buffer)
-        return AuthenticationSwitchRequest(method, salt)
+        buffer.skipBytes(bytes)
+        return AuthenticationSwitchRequest(method, seed)
     }
 }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt
index 94b14480..ae3eb915 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationMethod.kt
@@ -4,15 +4,19 @@ import java.nio.charset.Charset
 
 interface AuthenticationMethod {
 
-    fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray): ByteArray
+    fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray
 
     companion object {
-        val Native = "mysql_native_password"
-        val Old = "mysql_old_password"
+        const val CachingSha2 = "caching_sha2_password"
+        const val Native = "mysql_native_password"
+        const val Old = "mysql_old_password"
+        const val Sha256 = "sha256_password"
 
         val Availables = mapOf(
+            CachingSha2 to CachingSha2PasswordAuthentication,
             Native to MySQLNativePasswordAuthentication,
-            Old to OldPasswordAuthentication
+            Old to OldPasswordAuthentication,
+            Sha256 to Sha256PasswordAuthentication,
         )
     }
 }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationScrambler.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationScrambler.kt
new file mode 100644
index 00000000..7ff4afd9
--- /dev/null
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/AuthenticationScrambler.kt
@@ -0,0 +1,44 @@
+package com.github.jasync.sql.db.mysql.encoder.auth
+
+import com.github.jasync.sql.db.util.length
+import java.nio.charset.Charset
+import java.security.MessageDigest
+import kotlin.experimental.xor
+
+object AuthenticationScrambler {
+
+    fun scramble411(
+        algorithm: String,
+        password: String,
+        charset: Charset,
+        seed: ByteArray,
+        seedFirst: Boolean,
+    ): ByteArray {
+        val messageDigest = MessageDigest.getInstance(algorithm)
+        val initialDigest = messageDigest.digest(password.toByteArray(charset))
+
+        messageDigest.reset()
+
+        val finalDigest = messageDigest.digest(initialDigest)
+
+        messageDigest.reset()
+
+        if (seedFirst) {
+            messageDigest.update(seed)
+            messageDigest.update(finalDigest)
+        } else {
+            messageDigest.update(finalDigest)
+            messageDigest.update(seed)
+        }
+
+        val result = messageDigest.digest()
+        var counter = 0
+
+        while (counter < result.length) {
+            result[counter] = (result[counter] xor initialDigest[counter])
+            counter += 1
+        }
+
+        return result
+    }
+}
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/CachingSha2PasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/CachingSha2PasswordAuthentication.kt
new file mode 100644
index 00000000..1b7d1525
--- /dev/null
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/CachingSha2PasswordAuthentication.kt
@@ -0,0 +1,24 @@
+package com.github.jasync.sql.db.mysql.encoder.auth
+
+import java.nio.charset.Charset
+
+object CachingSha2PasswordAuthentication : AuthenticationMethod {
+
+    private val EmptyArray = ByteArray(0)
+
+    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
+        return if (password != null) {
+            if (seed != null) {
+                // Fast authentication mode. Requires seed, but not SSL.
+                AuthenticationScrambler.scramble411("SHA-256", password, charset, seed, false)
+            } else {
+                // Full authentication mode.
+                // Since this sends the plaintext password, SSL is required.
+                // Without SSL, the server always rejects the password.
+                Sha256PasswordAuthentication.generateAuthentication(charset, password, null)
+            }
+        } else {
+            EmptyArray
+        }
+    }
+}
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt
index 1379d263..d47018c5 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/MySQLNativePasswordAuthentication.kt
@@ -1,45 +1,18 @@
 package com.github.jasync.sql.db.mysql.encoder.auth
 
-import com.github.jasync.sql.db.util.length
 import java.nio.charset.Charset
-import java.security.MessageDigest
-import kotlin.experimental.xor
 
 object MySQLNativePasswordAuthentication : AuthenticationMethod {
 
-    val EmptyArray = ByteArray(0)
+    private val EmptyArray = ByteArray(0)
 
-    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray): ByteArray {
+    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
+        requireNotNull(seed) { "Seed should not be null" }
 
         return if (password != null) {
-            scramble411(charset, password, seed)
+            AuthenticationScrambler.scramble411("SHA-1", password, charset, seed, true)
         } else {
             EmptyArray
         }
     }
-
-    private fun scramble411(charset: Charset, password: String, seed: ByteArray): ByteArray {
-
-        val messageDigest = MessageDigest.getInstance("SHA-1")
-        val initialDigest = messageDigest.digest(password.toByteArray(charset))
-
-        messageDigest.reset()
-
-        val finalDigest = messageDigest.digest(initialDigest)
-
-        messageDigest.reset()
-
-        messageDigest.update(seed)
-        messageDigest.update(finalDigest)
-
-        val result = messageDigest.digest()
-        var counter = 0
-
-        while (counter < result.length) {
-            result[counter] = (result[counter] xor initialDigest[counter])
-            counter += 1
-        }
-
-        return result
-    }
 }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt
index ba87c646..9dafaea6 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/OldPasswordAuthentication.kt
@@ -7,12 +7,22 @@ import kotlin.math.floor
 @Suppress("RedundantExplicitType", "UNUSED_VALUE", "VARIABLE_WITH_REDUNDANT_INITIALIZER")
 object OldPasswordAuthentication : AuthenticationMethod {
 
-    val EmptyArray = ByteArray(0)
+    private val EmptyArray = ByteArray(0)
+
+    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
+        requireNotNull(seed) { "Seed should not be null" }
 
-    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray): ByteArray {
         return when {
-            password != null && password.isNotEmpty() -> {
-                newCrypt(charset, password, String(seed, charset))
+            !password.isNullOrEmpty() -> {
+                // The native authentication handshake will provide a 20-byte challenge.
+                // Use the first 8 bytes as the old password authentication challenge.
+                val challenge = if (seed.length == 20) {
+                    seed.copyOf(8)
+                } else {
+                    seed
+                }
+
+                newCrypt(charset, password, String(challenge, charset))
             }
             else -> EmptyArray
         }
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/Sha256PasswordAuthentication.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/Sha256PasswordAuthentication.kt
new file mode 100644
index 00000000..5ae0edca
--- /dev/null
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/encoder/auth/Sha256PasswordAuthentication.kt
@@ -0,0 +1,21 @@
+package com.github.jasync.sql.db.mysql.encoder.auth
+
+import com.github.jasync.sql.db.util.length
+import java.nio.charset.Charset
+
+// TODO: Implement public key encryption.
+object Sha256PasswordAuthentication : AuthenticationMethod {
+
+    private val EmptyArray = ByteArray(0)
+
+    override fun generateAuthentication(charset: Charset, password: String?, seed: ByteArray?): ByteArray {
+        return if (password != null) {
+            val bytes = password.toByteArray(charset)
+            val result = ByteArray(bytes.length + 1)
+            bytes.copyInto(result)
+            result
+        } else {
+            EmptyArray
+        }
+    }
+}
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthMoreDataMessage.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthMoreDataMessage.kt
new file mode 100644
index 00000000..dfeaa4de
--- /dev/null
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthMoreDataMessage.kt
@@ -0,0 +1,9 @@
+package com.github.jasync.sql.db.mysql.message.server
+
+data class AuthMoreDataMessage(
+    val data: Byte,
+) : ServerMessage(AuthMoreData) {
+    fun isSuccess(): Boolean {
+        return data == 3.toByte()
+    }
+}
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt
index 1ce19004..b06843dd 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/AuthenticationSwitchRequest.kt
@@ -2,5 +2,5 @@ package com.github.jasync.sql.db.mysql.message.server
 
 data class AuthenticationSwitchRequest(
     val method: String,
-    val seed: ByteArray
+    val seed: ByteArray?,
 ) : ServerMessage(ServerMessage.EOF)
diff --git a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/ServerMessage.kt b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/ServerMessage.kt
index 5df28459..8a2cf028 100644
--- a/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/ServerMessage.kt
+++ b/mysql-async/src/main/java/com/github/jasync/sql/db/mysql/message/server/ServerMessage.kt
@@ -8,6 +8,7 @@ abstract class ServerMessage(override val kind: Int) : KindedMessage {
         const val ServerProtocolVersion = 10
         const val Error = -1
         const val Ok = 0
+        const val AuthMoreData = 1
         const val EOF = -2
 
         // these messages don't actually exist
diff --git a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/AuthenticationSpec.kt b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/AuthenticationSpec.kt
new file mode 100644
index 00000000..7103420c
--- /dev/null
+++ b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/AuthenticationSpec.kt
@@ -0,0 +1,148 @@
+package com.github.jasync.sql.db.mysql
+
+import com.github.jasync.sql.db.Configuration
+import com.github.jasync.sql.db.SSLConfiguration
+import com.github.jasync.sql.db.SSLConfiguration.Mode
+import com.github.jasync.sql.db.invoke
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.Test
+import org.testcontainers.containers.MySQLContainer
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
+
+class AuthenticationSpec {
+
+    @Test
+    fun cachingSha2PasswordAuthentication() {
+        val container = createContainer("mysql:8.0.31")
+
+        withConnection(container, "root", "test") { connection ->
+            connection.sendQuery("CREATE USER 'user' IDENTIFIED WITH caching_sha2_password BY 'foo'").await()
+            connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user'").await()
+        }
+
+        // First connection without SSL fails because we need to perform full authentication.
+        assertThatThrownBy {
+            withConnection(container, "user", "foo") { /* Empty */ }
+        }.hasCauseInstanceOf(IllegalStateException::class.java)
+            .hasRootCauseMessage("Full authentication mode for caching_sha2_password requires SSL")
+
+        // Perform full authentication with SSL.
+        withConnection(container, "user", "foo", SSL_MODE) { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("caching_sha2_password")
+        }
+
+        // Perform fast authentication without SSL.
+        withConnection(container, "user", "foo") { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("caching_sha2_password")
+        }
+
+        container.stop()
+    }
+
+    @Test
+    fun oldPasswordAuthentication() {
+        val container = createContainer("mysql:5.6.51", "--skip-secure-auth")
+
+        withConnection(container, "root", "test") { connection ->
+            connection.sendQuery("CREATE USER 'user' IDENTIFIED WITH mysql_old_password").await()
+            connection.sendQuery("SET old_passwords = 1").await()
+            connection.sendQuery("SET PASSWORD FOR 'user' = PASSWORD('foo')").await()
+            connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user'").await()
+        }
+
+        withConnection(container, "user", "foo") { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("mysql_old_password")
+        }
+
+        container.stop()
+    }
+
+    @Test
+    fun nativePasswordAuthentication() {
+        val container = createContainer("mysql:8.0.31")
+
+        withConnection(container, "root", "test") { connection ->
+            connection.sendQuery("CREATE USER 'user' IDENTIFIED WITH mysql_native_password BY 'foo'").await()
+            connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user'").await()
+        }
+
+        withConnection(container, "user", "foo") { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("mysql_native_password")
+        }
+
+        container.stop()
+    }
+
+    @Test
+    fun sha256Authentication() {
+        val container = createContainer("mysql:5.7.32")
+
+        withConnection(container, "root", "test") { connection ->
+            connection.sendQuery("CREATE USER 'user' IDENTIFIED WITH sha256_password BY 'foo'").await()
+            connection.sendQuery("GRANT ALL PRIVILEGES ON *.* to 'user'").await()
+        }
+
+        withConnection(container, "user", "foo", SSL_MODE) { connection ->
+            val result = connection.sendQuery(QUERY_CURRENT_PLUGIN).await()
+            assertThat(result.rows).hasSize(1)
+            assertThat(result.rows[0]("plugin")).isEqualTo("sha256_password")
+        }
+
+        container.stop()
+    }
+
+    private fun createContainer(imageName: String, command: String? = null): MySQLContainer<*> {
+        val container = MySQLContainer(imageName)
+            .withUsername("root")
+            .withPassword("test")
+
+        if (command != null) {
+            container.withCommand(command)
+        }
+
+        container.start()
+        return container
+    }
+
+    private fun <T> withConnection(
+        container: MySQLContainer<*>,
+        username: String,
+        password: String,
+        sslConfiguration: SSLConfiguration = SSLConfiguration(Mode.Disable),
+        fn: (MySQLConnection) -> T,
+    ): T {
+        val configuration = Configuration(
+            username = username,
+            password = password,
+            port = container.firstMappedPort,
+            ssl = sslConfiguration,
+        )
+
+        val connection = MySQLConnection(configuration)
+        connection.connect().await()
+
+        val result = fn(connection)
+        connection.close().await()
+
+        return result
+    }
+
+    private fun <T> CompletableFuture<T>.await(): T {
+        return this.get(10, TimeUnit.SECONDS)
+    }
+
+    private companion object {
+        const val QUERY_CURRENT_PLUGIN = "SELECT plugin FROM mysql.user WHERE CURRENT_USER() = CONCAT(user, '@', host);"
+        val SSL_MODE = SSLConfiguration(Mode.Prefer)
+    }
+}
diff --git a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java
index 1f11b0a2..0a761f7d 100644
--- a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java
+++ b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/ContainerHelper.java
@@ -5,6 +5,9 @@
 import com.github.jasync.sql.db.QueryResult;
 import com.github.jasync.sql.db.ResultSet;
 import com.github.jasync.sql.db.RowData;
+import com.github.jasync.sql.db.SSLConfiguration;
+import com.github.jasync.sql.db.SSLConfiguration.Mode;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.BindMode;
@@ -31,25 +34,30 @@ public static Integer getPort() {
         return defaultConfiguration.getPort();
     }
 
+    private static final SSLConfiguration sslConfiguration =
+        new SSLConfiguration(Mode.Prefer, null, null, null);
+
     /**
      * default config is a local instance already running on port 33306 (i.e. a docker mysql)
      */
     public static Configuration defaultConfiguration = new Configuration(
-            "mysql_async",
-            "localhost",
-            33306,
-            "root",
-            "mysql_async_tests");
+        "mysql_async",
+        "localhost",
+        33306,
+        "root",
+        "mysql_async_tests",
+        sslConfiguration);
 
     /**
      * config for container.
      */
     private static Configuration rootConfiguration = new Configuration(
-            "root",
-            "localhost",
-            33306,
-            "test",
-            "mysql_async_tests");
+        "root",
+        "localhost",
+        33306,
+        "test",
+        "mysql_async_tests",
+        sslConfiguration);
 
     private static boolean isLocalMySQLRunning() {
         try {
@@ -82,7 +90,7 @@ private static boolean isLocalMySQLRunning() {
 
     private static void startMySQLDocker() throws IOException {
         if (mysql == null) {
-            mysql = new MySQLContainer("mysql:5.7.32") {
+            mysql = new MySQLContainer("mysql:8.0.31") {
                 @Override
                 protected void configure() {
                     super.configure();
@@ -112,14 +120,14 @@ protected void configure() {
         if (!mysql.isRunning()) {
             mysql.start();
         }
-        defaultConfiguration = new Configuration("mysql_async", "localhost", mysql.getFirstMappedPort(), "root", "mysql_async_tests");
-        rootConfiguration = new Configuration("root", "localhost", mysql.getFirstMappedPort(), "test", "mysql_async_tests");
+        defaultConfiguration = new Configuration("mysql_async", "localhost", mysql.getFirstMappedPort(), "root", "mysql_async_tests", sslConfiguration);
+        rootConfiguration = new Configuration("root", "localhost", mysql.getFirstMappedPort(), "test", "mysql_async_tests", sslConfiguration);
         logger.info("Using test container instance {}", defaultConfiguration);
     }
 
     private static void configureDatabase() throws Exception {
         Connection connection = new MySQLConnection(rootConfiguration).connect().get(1, TimeUnit.SECONDS);
-        connection.sendQuery("GRANT ALL PRIVILEGES ON *.* TO 'mysql_async'@'%' IDENTIFIED BY 'root' WITH GRANT OPTION;").get(1, TimeUnit.SECONDS);
+        connection.sendQuery("GRANT ALL PRIVILEGES ON *.* TO 'mysql_async'@'%' WITH GRANT OPTION;").get(1, TimeUnit.SECONDS);
         QueryResult r = connection.sendQuery("select count(*) as cnt  from mysql.user where user = 'mysql_async_nopw';").get(1, TimeUnit.SECONDS);
         r.getRows();
         if (r.getRows().size() > 0) {
diff --git a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/StoredProceduresSpec.kt b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/StoredProceduresSpec.kt
index 4ca74e67..72e449fe 100644
--- a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/StoredProceduresSpec.kt
+++ b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/StoredProceduresSpec.kt
@@ -96,7 +96,7 @@ class StoredProceduresSpec : ConnectionHelper() {
             ).get()
             val rows = connection.sendQuery(
                 """
-                  SELECT routine_name FROM INFORMATION_SCHEMA.ROUTINES WHERE routine_name="remTest"
+                  SELECT routine_name as routine_name FROM INFORMATION_SCHEMA.ROUTINES WHERE routine_name="remTest"
                 """
             ).get().rows
 
diff --git a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/TransactionSpec.kt b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/TransactionSpec.kt
index 602d08cf..258f3044 100644
--- a/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/TransactionSpec.kt
+++ b/mysql-async/src/test/java/com/github/jasync/sql/db/mysql/TransactionSpec.kt
@@ -68,7 +68,7 @@ class TransactionSpec : ConnectionHelper() {
             } as MySQLException
 
             assertThat(e.errorMessage.errorCode).isEqualTo(1062)
-            assertThat(e.errorMessage.errorMessage).isEqualTo("Duplicate entry '1' for key 'PRIMARY'")
+            assertThat(e.errorMessage.errorMessage).matches("Duplicate entry '1' for key '(users\\.)?PRIMARY'")
 
             val result = executePreparedStatement(connection, this.select).rows
             assertThat(result.size).isEqualTo(1)
diff --git a/mysql-async/src/test/resources/update-config.sh b/mysql-async/src/test/resources/update-config.sh
index 6890a99a..d56e28ec 100644
--- a/mysql-async/src/test/resources/update-config.sh
+++ b/mysql-async/src/test/resources/update-config.sh
@@ -2,7 +2,8 @@
 cp -f /docker-entrypoint-initdb.d/ca.pem /var/lib/mysql/ca.pem
 cp -f /docker-entrypoint-initdb.d/server-cert.pem /var/lib/mysql/server-cert.pem
 cp -f /docker-entrypoint-initdb.d/server-key.pem /var/lib/mysql/server-key.pem
-cat /etc/mysql/my.cnf
+cat /etc/my.cnf || true
+cat /etc/mysql/my.cnf || true
 ls -lah /var/lib/mysql
-ls -lah /etc/mysql/conf.d/
-ls -lah /etc/mysql/mysql.conf.d/
+ls -lah /etc/mysql/conf.d/ || true
+ls -lah /etc/mysql/mysql.conf.d/ || true
diff --git a/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java b/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java
index 99322772..1ca5dadf 100644
--- a/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java
+++ b/r2dbc-mysql/src/test/java/com/github/jasync/r2dbc/mysql/integ/R2dbcContainerHelper.java
@@ -4,6 +4,8 @@
 import com.github.jasync.sql.db.Connection;
 import com.github.jasync.sql.db.QueryResult;
 import com.github.jasync.sql.db.RowData;
+import com.github.jasync.sql.db.SSLConfiguration;
+import com.github.jasync.sql.db.SSLConfiguration.Mode;
 import com.github.jasync.sql.db.mysql.MySQLConnection;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -25,25 +27,30 @@ public static Integer getPort() {
         return defaultConfiguration.getPort();
     }
 
+    private static final SSLConfiguration sslConfiguration =
+        new SSLConfiguration(Mode.Prefer, null, null, null);
+
     /**
      * default config is a local instance already running on port 33306 (i.e. a docker mysql)
      */
     public static Configuration defaultConfiguration = new Configuration(
-            "mysql_async",
-            "localhost",
-            33306,
-            "root",
-            "mysql_async_tests");
+        "mysql_async",
+        "localhost",
+        33306,
+        "root",
+        "mysql_async_tests",
+        sslConfiguration);
 
     /**
      * config for container.
      */
     private static Configuration rootConfiguration = new Configuration(
-            "root",
-            "localhost",
-            33306,
-            "test",
-            "mysql_async_tests");
+        "root",
+        "localhost",
+        33306,
+        "test",
+        "mysql_async_tests",
+        sslConfiguration);
 
     private static boolean isLocalMySQLRunning() {
         try {
@@ -60,7 +67,7 @@ private static boolean isLocalMySQLRunning() {
 
     private static void startMySQLDocker() {
         if (mysql == null) {
-            mysql = new MySQLContainer("mysql:5.7.32") {
+            mysql = new MySQLContainer("mysql:8.0.31") {
                 @Override
                 protected void configure() {
                     super.configure();
@@ -82,14 +89,14 @@ protected void configure() {
         if (!mysql.isRunning()) {
             mysql.start();
         }
-        defaultConfiguration = new Configuration("mysql_async", "localhost", mysql.getFirstMappedPort(), "root", "mysql_async_tests");
-        rootConfiguration = new Configuration("root", "localhost", mysql.getFirstMappedPort(), "test", "mysql_async_tests");
+        defaultConfiguration = new Configuration("mysql_async", "localhost", mysql.getFirstMappedPort(), "root", "mysql_async_tests", sslConfiguration);
+        rootConfiguration = new Configuration("root", "localhost", mysql.getFirstMappedPort(), "test", "mysql_async_tests", sslConfiguration);
         logger.info("Using test container instance {}", defaultConfiguration);
     }
 
     private static void configureDatabase() throws Exception {
         Connection connection = new MySQLConnection(rootConfiguration).connect().get(1, TimeUnit.SECONDS);
-        connection.sendQuery("GRANT ALL PRIVILEGES ON *.* TO 'mysql_async'@'%' IDENTIFIED BY 'root' WITH GRANT OPTION;").get(1, TimeUnit.SECONDS);
+        connection.sendQuery("GRANT ALL PRIVILEGES ON *.* TO 'mysql_async'@'%' WITH GRANT OPTION;").get(1, TimeUnit.SECONDS);
         QueryResult r = connection.sendQuery("select count(*) as cnt  from mysql.user where user = 'mysql_async_nopw';").get(1, TimeUnit.SECONDS);
         r.getRows();
         if (r.getRows().size() > 0) {
diff --git a/r2dbc-mysql/src/test/resources/update-config.sh b/r2dbc-mysql/src/test/resources/update-config.sh
index 6890a99a..d56e28ec 100644
--- a/r2dbc-mysql/src/test/resources/update-config.sh
+++ b/r2dbc-mysql/src/test/resources/update-config.sh
@@ -2,7 +2,8 @@
 cp -f /docker-entrypoint-initdb.d/ca.pem /var/lib/mysql/ca.pem
 cp -f /docker-entrypoint-initdb.d/server-cert.pem /var/lib/mysql/server-cert.pem
 cp -f /docker-entrypoint-initdb.d/server-key.pem /var/lib/mysql/server-key.pem
-cat /etc/mysql/my.cnf
+cat /etc/my.cnf || true
+cat /etc/mysql/my.cnf || true
 ls -lah /var/lib/mysql
-ls -lah /etc/mysql/conf.d/
-ls -lah /etc/mysql/mysql.conf.d/
+ls -lah /etc/mysql/conf.d/ || true
+ls -lah /etc/mysql/mysql.conf.d/ || true