diff --git a/src/main/scala/io/lenses/connect/secrets/config/VaultProviderConfig.scala b/src/main/scala/io/lenses/connect/secrets/config/VaultProviderConfig.scala
index 3df6828..d94d80b 100644
--- a/src/main/scala/io/lenses/connect/secrets/config/VaultProviderConfig.scala
+++ b/src/main/scala/io/lenses/connect/secrets/config/VaultProviderConfig.scala
@@ -27,6 +27,7 @@ object VaultProviderConfig {
   val VAULT_CLIENT_PEM: String = "vault.client.pem"
   val VAULT_PEM: String = "vault.pem"
   val VAULT_ENGINE_VERSION = "vault.engine.version"
+  val VAULT_SECRET_TTL_DEFAULT: String = "vault.secret.ttl.default"
   val AUTH_METHOD: String = "vault.auth.method"
 
   val VAULT_TRUSTSTORE_LOC: String =
@@ -137,6 +138,13 @@ object VaultProviderConfig {
       Importance.HIGH,
       "KV Secrets Engine version of the Vault server instance. Defaults to 2"
     )
+    .define(
+      VAULT_SECRET_TTL_DEFAULT,
+      Type.INT,
+      0,
+      Importance.MEDIUM,
+      "Default TTL for secrets returned from Vault. Defaults to 0 to not break existing configurations. Work around for the lack of `lease_duration` values returned from the Vault KV2 engine: https://github.com/hashicorp/vault/issues/6274"
+    )
     // auth mode
     .define(
       AUTH_METHOD,
diff --git a/src/main/scala/io/lenses/connect/secrets/config/VaultProviderSettings.scala b/src/main/scala/io/lenses/connect/secrets/config/VaultProviderSettings.scala
index cdf4fec..f1a9c6d 100644
--- a/src/main/scala/io/lenses/connect/secrets/config/VaultProviderSettings.scala
+++ b/src/main/scala/io/lenses/connect/secrets/config/VaultProviderSettings.scala
@@ -46,6 +46,7 @@ case class VaultSettings(
     pem: String,
     clientPem: String,
     engineVersion: Int = 2,
+    secretTtlDefault: Int = 0,
     appRole: Option[AppRole],
     awsIam: Option[AwsIam],
     gcp: Option[Gcp],
@@ -72,6 +73,7 @@ object VaultSettings extends StrictLogging {
     val pem = config.getString(VaultProviderConfig.VAULT_PEM)
     val clientPem = config.getString(VaultProviderConfig.VAULT_CLIENT_PEM)
     val engineVersion = config.getInt(VaultProviderConfig.VAULT_ENGINE_VERSION)
+    val secretTtlDefault = config.getInt(VaultProviderConfig.VAULT_SECRET_TTL_DEFAULT)
 
     val authMode = VaultAuthMethod.withNameOpt(
       config.getString(VaultProviderConfig.AUTH_METHOD).toUpperCase
@@ -121,6 +123,7 @@ object VaultSettings extends StrictLogging {
       pem = pem,
       clientPem = clientPem,
       engineVersion = engineVersion,
+      secretTtlDefault = secretTtlDefault,
       appRole = appRole,
       awsIam = awsIam,
       gcp = gcp,
diff --git a/src/main/scala/io/lenses/connect/secrets/providers/VaultSecretProvider.scala b/src/main/scala/io/lenses/connect/secrets/providers/VaultSecretProvider.scala
index 3d4f9a0..e5c5865 100644
--- a/src/main/scala/io/lenses/connect/secrets/providers/VaultSecretProvider.scala
+++ b/src/main/scala/io/lenses/connect/secrets/providers/VaultSecretProvider.scala
@@ -6,7 +6,6 @@
 
 package io.lenses.connect.secrets.providers
 
-import java.nio.file.FileSystems
 import java.nio.file.Paths
 import java.time.OffsetDateTime
 import java.util
@@ -15,6 +14,7 @@ import _root_.io.lenses.connect.secrets.config.VaultProviderConfig
 import _root_.io.lenses.connect.secrets.config.VaultSettings
 import _root_.io.lenses.connect.secrets.connect._
 import com.bettercloud.vault.Vault
+import com.bettercloud.vault.response.LogicalResponse
 import io.lenses.connect.secrets.async.AsyncFunctionLoop
 import io.lenses.connect.secrets.io.FileWriterOnce
 import io.lenses.connect.secrets.utils.EncodingAndId
@@ -30,7 +30,6 @@ import scala.util.Try
 
 class VaultSecretProvider() extends ConfigProvider with VaultHelper {
 
-  private val separator: String = FileSystems.getDefault.getSeparator
   private var settings: VaultSettings = _
   private var vaultClient: Option[Vault] = None
   private var tokenRenewal: Option[AsyncFunctionLoop] = None
@@ -122,7 +121,7 @@ class VaultSecretProvider() extends ConfigProvider with VaultHelper {
 
     logger.info(s"Looking up value at [$path]")
     val fileWriter = new FileWriterOnce(Paths.get(settings.fileDir, path))
-    Try(vaultClient.get.logical().read(path)) match {
+    Try(readSecretFromVault(path)) match {
       case Success(response) =>
         if (response.getRestResponse.getStatus != 200) {
           throw new ConnectException(
@@ -133,8 +132,15 @@ class VaultSecretProvider() extends ConfigProvider with VaultHelper {
           )
         }
 
-        val ttl = Option(vaultClient.get.logical().read(path).getLeaseDuration) match {
-          case Some(duration) => Some(now.plusSeconds(duration))
+        val ttl = Option(response.getLeaseDuration) match {
+          case Some(duration) =>
+            if (duration != 0) {
+              Some(now.plusSeconds(duration))
+            } else if (settings.secretTtlDefault > 0) {
+              Some(now.plusSeconds(settings.secretTtlDefault))
+            } else {
+              None
+            }
           case None => None
         }
 
@@ -162,4 +168,20 @@ class VaultSecretProvider() extends ConfigProvider with VaultHelper {
         )
     }
   }
+
+  private def readSecretFromVault(path: String): LogicalResponse = {
+    // Check if the path looks like it's referencing the vault database engine. If so, use the `database()` api of the
+    // vault client to obtain this secret.  Otherwise, fall back to the default key-vaule engine.
+    if (path.startsWith("database/creds/")) {
+      val parts = path.split("/")
+
+      if (parts.length == 3) {
+        vaultClient.get.database().creds(parts(2))
+      } else {
+        throw new Exception("Database path is invalid. Path must be in the form 'database/creds/<role_name>'")
+      }
+    } else {
+      vaultClient.get.logical().read(path)
+    }
+  }
 }