Repository: microsoft/mssql-jdbc
Pull request #1640 (2021-08-13 16:18:43)

Fix conversion of LocalDateTime and LocalTime to String in Bulk Copy
When writing a LocalDateTime with bulk copy, the driver may return an error when second-of-minute and nano-of-second is 0:

```
com.microsoft.sqlserver.jdbc.SQLServerException: Conversion failed when converting date and/or time from character string. Msg 241, Level 16, State 1, Conversion failed when converting date and/or time from character string.
```

When time units to the right of the minute are all 0's, the LocalDateTime.toString() will simplify the string format to **uuuu-MM-dd'T'HH:mm**, which SQL Server is not able to process due to the missing seconds. This PR changes the string conversion to use the DateTimeFormatter and ensures that it includes the necessary padding.

For example:
```
LocalDateTime                  = 2021-08-13 00:00:00

Current - toString()           = 2021-08-13T00:00              // Returns error
New     - DateTimeFormatter    = 2021-08-13T00:00:00           // It works!
```

Note that this is very similar to PR #1485.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index da22692c..5fe4682b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,21 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 
+## [9.5.0] Preview Release
+### Added
+- Idle Connection Resiliency Feature [1669](https://github.com/microsoft/mssql-jdbc/pull/1669)
+- Fix for Bulkcopy multi byte characters in char/vchar columns [1671](https://github.com/microsoft/mssql-jdbc/pull/1671)
+- Java 17 support [1676](https://github.com/microsoft/mssql-jdbc/pull/1676)
+- Added logging when deriving realm [1672](https://github.com/microsoft/mssql-jdbc/pull/1672)
+- Added check for closed statement to registerColumnEncryptionKeyStoreProvidersOnStatement [1644](https://github.com/microsoft/mssql-jdbc/pull/1644)
+- Added 42108 and 42109 to transient error list [1643](https://github.com/microsoft/mssql-jdbc/pull/1643)
+### Changed
+- Updated MD4 based on Bouncy Castle implementation [1665](https://github.com/microsoft/mssql-jdbc/pull/1665)
+- Updated mssql-jdbc_auth_LICENSE [1663](https://github.com/microsoft/mssql-jdbc/pull/1663)
+### Fixed issues
+- Fixed and issue where Statement.closeOnCompletion() would cause a Null Pointer Exeception [1639](https://github.com/microsoft/mssql-jdbc/pull/1639)
+
+
 ## [9.4.0] Stable Release
 ### Added
 - Added JAVA 16 support [#1579](https://github.com/microsoft/mssql-jdbc/pull/1579)
diff --git a/README.md b/README.md
index 19e1db87..a51b67c4 100644
--- a/README.md
+++ b/README.md
@@ -47,14 +47,14 @@ To build the jar files, you must use minimum version of Java 11 with Maven. You
 * Maven:
 	1. If you have not already done so, add the environment variable `mssql_jdbc_test_connection_properties` in your system with the connection properties for your SQL Server or SQL DB instance.
 	2. Run one of the commands below to build a JRE 11 and newer versions compatible jar or JRE 8 compatible jar in the `\target` directory. 
-        * Run `mvn install -Pjre16`. This creates JRE 16 compatible jar in `\target` directory which is JDBC 4.3 compliant (Build with JDK 16+).
+        * Run `mvn install -Pjre17`. This creates JRE 17 compatible jar in `\target` directory which is JDBC 4.3 compliant (Build with JDK 17+).
         * Run `mvn install -Pjre11`. This creates JRE 11 compatible jar in `\target` directory which is JDBC 4.3 compliant (Build with JDK 11+).
         * Run `mvn install -Pjre8`. This creates JRE 8 compatible jar in `\target` directory which is JDBC 4.2 compliant (Build with JDK 11+).
 
 * Gradle:
 	1. If you have not already done so, add the environment variable `mssql_jdbc_test_connection_properties` in your system with the connection properties for your SQL Server or SQL DB instance.
 	2. Run one of the commands below to build a JRE 11 and newer versions compatible jar or JRE 8 compatible jar in the `\build\libs` directory. 
-        * Run `gradle build -PbuildProfile=jre16`. This creates JRE 16 compatible jar in `\build\libs` directory which is JDBC 4.3 compliant (Build with JDK 16+).
+        * Run `gradle build -PbuildProfile=jre17`. This creates JRE 17 compatible jar in `\build\libs` directory which is JDBC 4.3 compliant (Build with JDK 17+).
         * Run `gradle build -PbuildProfile=jre11`. This creates JRE 11 compatible jar in `\build\libs` directory which is JDBC 4.3 compliant (Build with JDK 11+).
         * Run `gradle build -PbuildProfile=jre8`. This creates JRE 8 compatible jar in `\build\libs` directory which is JDBC 4.2 compliant (Build with JDK 11+).
 
@@ -80,7 +80,7 @@ We're now on the Maven Central Repository. Add the following to your POM file to
 <dependency>
 	<groupId>com.microsoft.sqlserver</groupId>
 	<artifactId>mssql-jdbc</artifactId>
-	<version>9.4.0.jre16</version>
+	<version>9.5.0.jre17</version>
 </dependency>
 ```
 The driver can be downloaded from the [Microsoft Download Center](https://go.microsoft.com/fwlink/?linkid=2168495).
@@ -91,7 +91,7 @@ To get the latest preview version of the driver, add the following to your POM f
 <dependency>
 	<groupId>com.microsoft.sqlserver</groupId>
 	<artifactId>mssql-jdbc</artifactId>
-	<version>9.4.0.jre16</version>
+	<version>9.5.0.jre17</version>
 </dependency>
 ```
 
@@ -126,7 +126,7 @@ Projects that require either of the two features need to explicitly declare the
 <dependency>
 	<groupId>com.microsoft.sqlserver</groupId>
 	<artifactId>mssql-jdbc</artifactId>
-	<version>9.4.0.jre16</version>
+	<version>9.5.0.jre17</version>
 	<scope>compile</scope>
 </dependency>
 
@@ -144,7 +144,7 @@ Projects that require either of the two features need to explicitly declare the
 <dependency>
 	<groupId>com.microsoft.sqlserver</groupId>
 	<artifactId>mssql-jdbc</artifactId>
-	<version>9.4.0.jre16</version>
+	<version>9.5.0.jre17</version>
 	<scope>compile</scope>
 </dependency>
 
@@ -171,7 +171,7 @@ When setting 'useFmtOnly' property to 'true' for establishing a connection or cr
 <dependency>
 	<groupId>com.microsoft.sqlserver</groupId>
 	<artifactId>mssql-jdbc</artifactId>
-	<version>9.4.0.jre16</version>
+	<version>9.5.0.jre17</version>
 </dependency>
 
 <dependency>
@@ -211,7 +211,7 @@ Preview releases happen approximately monthly between stable releases. This give
 You can see what is going into a future release by monitoring [Milestones](https://github.com/Microsoft/mssql-jdbc/milestones) in the repository.
 
 ### Version conventions
-Starting with 6.0, stable versions have an even minor version. For example, 6.0, 6.2, 6.4, 7.0, 7.2, 7.4, 8.2, 8.4, 9.2, 9.4. Preview versions have an odd minor version. For example, 6.1, 6.3, 6.5, 7.1, 7.3, 8.1, 9.1 and so on.
+Starting with 6.0, stable versions have an even minor version. For example, 6.0, 6.2, 6.4, 7.0, 7.2, 7.4, 8.2, 8.4, 9.2, 9.4, 9.5. Preview versions have an odd minor version. For example, 6.1, 6.3, 6.5, 7.1, 7.3, 8.1, 9.1 and so on.
 
 ## Contributors 
 Special thanks to everyone who has contributed to the project. 
diff --git a/azure-pipelines.yml b/azure-pipelines.yml
deleted file mode 100644
index e6c8c6ac..00000000
--- a/azure-pipelines.yml
+++ /dev/null
@@ -1,71 +0,0 @@
-# Microsoft JDBC Driver for SQL Server CI build triggers tests against below SQL Servers:
-#   - SQL Server 2019
-#   - SQL Server 2012
-jobs:
-- job: "CI_Build"
-  pool:
-    name: 'JDBC Build Pool'
-    demands: maven
-  strategy:
-    matrix:
-      SQL-2019:
-        Target_SQL: 'HGS-2k19-01'
-        Ex_Groups: 'xSQLv15,MSI,clientCertAuth,fedAuth'
-      SQL-2012:
-        Target_SQL: 'SQL-2K12-SP3-1'
-        Ex_Groups: 'xSQLv12,MSI,fedAuth'
-    maxParallel: 2
-  steps:
-  - powershell: |
-      mkdir AE_Certificates
-      cd AE_Certificates
-      $cert = New-SelfSignedCertificate -dns "AlwaysEncryptedCert" -CertStoreLocation Cert:CurrentUser\My
-      $pwd = ConvertTo-SecureString -String $(certPass) -Force -AsPlainText
-      $path = 'cert:\CurrentUser\My\' + $cert.thumbprint
-      $certificate = Export-PfxCertificate -cert $path -FilePath cert.pfx -Password $pwd 
-      Get-ChildItem -path cert:\CurrentUser\My > certificate.txt
-      keytool -importkeystore -srckeystore cert.pfx -srcstoretype pkcs12 -destkeystore clientcert.jks -deststoretype pkcs12 -srcstorepass $(certPass) -deststorepass $(certPass)
-      keytool -list -v -keystore clientcert.jks -storepass $(certPass) > JavaKeyStoreBase.txt
-      Get-Content .\JavaKeyStoreBase.txt | Set-Content -Encoding utf8 JavaKeyStore.txt
-      Remove-Item –path .\JavaKeyStoreBase.txt
-    displayName: 'PowerShell Script'
-  - task: DownloadSecureFile@1
-    name: pkcs12_truststore
-    displayName: 'Download PKCS12 truststore file'
-    inputs:
-      secureFile: 'pkcs12_truststore'
-  - task: DownloadSecureFile@1
-    name: authDLL
-    displayName: 'Download mssql-jdbc_auth DLL'
-    inputs:
-      secureFile: 'mssql-jdbc_auth-9.5.0.x64-preview.dll'
-  - task: Maven@3
-    displayName: 'Maven build jre16'
-    inputs:
-      mavenPomFile: 'pom.xml'
-      goals: 'clean dependency:purge-local-repository -DdllPath=$(Agent.TempDirectory) -Dmssql_jdbc_test_connection_properties=jdbc:sqlserver://$(Target_SQL)$(server_domain);$(database);$(user);$(password); install -Pjre16 -DuserNTLM=$(userNTLM) -DpasswordNTLM=$(passwordNTLM) -DdomainNTLM=$(domainNTLM) -DexcludedGroups=$(Ex_Groups) -Dpkcs12_truststore_password=$(pkcs12_truststore_password) -Dpkcs12_truststore=$(pkcs12_truststore.secureFilePath)
--DapplicationClientID=$(applicationClientID) -DapplicationKey=$(applicationKey) -DtenantID=$(tenantID) -DkeyID=$(keyID) -DwindowsKeyPath=$(windowsKeyPath) -DenclaveAttestationUrl=$(enclaveAttestationUrl) -DenclaveAttestationProtocol=$(enclaveAttestationProtocol) -DenclaveServer=$(enclaveServer)'
-      testResultsFiles: '**/TEST-*.xml'
-      testRunTitle: 'Maven build jre16'
-      javaHomeOption: Path
-      jdkDirectory: $(JDK16)
-  - task: Maven@3
-    displayName: 'Maven build jre11'
-    inputs:
-      mavenPomFile: 'pom.xml'
-      goals: 'clean dependency:purge-local-repository -DdllPath=$(Agent.TempDirectory) -Dmssql_jdbc_test_connection_properties=jdbc:sqlserver://$(Target_SQL)$(server_domain);$(database);$(user);$(password); install -Pjre11 -DuserNTLM=$(userNTLM) -DpasswordNTLM=$(passwordNTLM) -DdomainNTLM=$(domainNTLM) -DexcludedGroups=$(Ex_Groups) -Dpkcs12_truststore_password=$(pkcs12_truststore_password) -Dpkcs12_truststore=$(pkcs12_truststore.secureFilePath)
--DapplicationClientID=$(applicationClientID) -DapplicationKey=$(applicationKey) -DtenantID=$(tenantID) -DkeyID=$(keyID) -DwindowsKeyPath=$(windowsKeyPath) -DenclaveAttestationUrl=$(enclaveAttestationUrl) -DenclaveAttestationProtocol=$(enclaveAttestationProtocol) -DenclaveServer=$(enclaveServer)'
-      testResultsFiles: '**/TEST-*.xml'
-      testRunTitle: 'Maven build jre11'
-      javaHomeOption: Path
-      jdkDirectory: $(JDK16)
-  - task: Maven@3
-    displayName: 'Maven build jre8'
-    inputs:
-      mavenPomFile: 'pom.xml'
-      goals: 'clean dependency:purge-local-repository -DdllPath=$(Agent.TempDirectory) -Dmssql_jdbc_test_connection_properties=jdbc:sqlserver://$(Target_SQL)$(server_domain);$(database);$(user);$(password); install -Pjre8 -DuserNTLM=$(userNTLM) -DpasswordNTLM=$(passwordNTLM) -DdomainNTLM=$(domainNTLM) -DexcludedGroups=$(Ex_Groups) -Dpkcs12_truststore_password=$(pkcs12_truststore_password) -Dpkcs12_truststore=$(pkcs12_truststore.secureFilePath) 
--DapplicationClientID=$(applicationClientID) -DapplicationKey=$(applicationKey) -DtenantID=$(tenantID) -DkeyID=$(keyID) -DwindowsKeyPath=$(windowsKeyPath) -DenclaveAttestationUrl=$(enclaveAttestationUrl) -DenclaveAttestationProtocol=$(enclaveAttestationProtocol) -DenclaveServer=$(enclaveServer)'
-      testResultsFiles: '**/TEST-*.xml'
-      testRunTitle: 'Maven build jre8'
-      javaHomeOption: Path
-      jdkDirectory: $(JDK16)
diff --git a/build.gradle b/build.gradle
index 2c631ddb..a33bb3c5 100644
--- a/build.gradle
+++ b/build.gradle
@@ -3,10 +3,7 @@
  ****************************************************************
  * Instruction for Building JDBC Driver:
  * For building particular version of the driver, use commands:
- * jre16 - - PS> gradle build
-			 PS> gradle build -PbuildProfile=jre16
- * jre11 - - PS> gradle build -PbuildProfile=jre11
- * jre8  - - PS> gradle build -PbuildProfile=jre8
+ * jreXX - - PS> gradle build -PbuildProfile=jreXX
  *
  * For Excluding Groups in command line:
  * PS> gradle build -PbuildProfile=jre11 "-PexcludedGroups=['xSQLv15','xGradle']"
@@ -35,17 +32,17 @@ test {
     }
 }
 
-if (!hasProperty('buildProfile') || (hasProperty('buildProfile') && buildProfile == "jre16")){
+if (!hasProperty('buildProfile') || (hasProperty('buildProfile') && buildProfile == "jre17")){
 	
-	jreVersion = "jre16"
+	jreVersion = "jre17"
 	excludedFile = 'com/microsoft/sqlserver/jdbc/SQLServerJdbc42.java'
 	jar {
 		manifest {
 			attributes 'Automatic-Module-Name': 'com.microsoft.sqlserver.jdbc'
 		}
 	}
-	sourceCompatibility = 16
-	targetCompatibility = 16
+	sourceCompatibility = 17
+	targetCompatibility = 17
 }
 
 if (hasProperty('buildProfile') && buildProfile == "jre11"){
diff --git a/issue_template.md b/issue_template.md
index e6975c0c..216a2c5a 100644
--- a/issue_template.md
+++ b/issue_template.md
@@ -1,5 +1,5 @@
 ## Driver version
-<!--- Provide the JDBC driver version (e.g. 7.0.0). -->
+<!--- Provide the JDBC driver version (e.g. 9.4.0). -->
 
 ## SQL Server version
 <!--- Provide the output of executing `SELECT @@VERSION` on your target SQL Server. -->
@@ -21,7 +21,7 @@
 4. Any other details that can be helpful:
 
 ## JDBC trace logs
-<!--- Provide the JDBC trace logs. Instructions can be found here: https://docs.microsoft.com/en-us/sql/connect/jdbc/tracing-driver-operation?view=sql-server-2017 -->
+<!--- Provide the JDBC trace logs. Instructions can be found here: https://docs.microsoft.com/sql/connect/jdbc/tracing-driver-operation -->
 
 ## Reproduction code
 <!--- Provide standalone JAVA code that reproduces the issue, preferably raw JDBC code. -->
diff --git a/mssql-jdbc_auth_LICENSE b/mssql-jdbc_auth_LICENSE
index e180c77a..526a5582 100644
--- a/mssql-jdbc_auth_LICENSE
+++ b/mssql-jdbc_auth_LICENSE
@@ -9,7 +9,7 @@ These license terms are an agreement between you and Microsoft Corporation (or o
 
 2. DISTRIBUTABLE CODE. The software may contain code you are permitted to distribute (i.e. make available for third parties) in applications you develop, as described in this Section. 
 	a) Distribution Rights. The code and test files described below are distributable if included with the software.
-		i. REDIST.TXT Files. You may copy and distribute the object code form of code listed on the REDIST list in the software, if any, or listed at REDIST;
+		i. REDIST.TXT Files. You may copy and distribute the object code form of code listed on the REDIST list in the software, if any, or listed at REDIST (https://aka.ms/jdbceularedist);
 		ii. Image Library. You may copy and distribute images, graphics, and animations in the Image Library as described in the software documentation;
 		iii. Sample Code, Templates, and Styles. You may copy, modify, and distribute the source and object code form of code marked as “sample”, “template”, “simple styles”, and “sketch styles”; and
 		iv. Third Party Distribution. You may permit distributors of your applications to copy and distribute any of this distributable code you elect to distribute with your applications.
diff --git a/pom.xml b/pom.xml
index 44cf3236..1693e75b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -41,22 +41,21 @@
 
 	<properties>
 		<!-- Allowed values for excluded Groups here - - - - - - - - - - - - - 
-			xJDBC42 - - - - - - For tests not compatible with JDBC 42 Specifications 
-			xGradle - - - - - - For tests not compatible with Gradle Script - - - - - 
-			xSQLv12 - - - - - - For tests not compatible with SQL Server 2008 R2 - 2014 
-			xSQLv14 - - - - - - For tests not compatible with SQL Server 2016 - 2017 
-			xSQLv15 - - - - - - For tests not compatible with SQL Server 2019 - - - - 
-			xAzureSQLDB - - - - For tests not compatible with Azure SQL Database - - 
-			xAzureSQLDW - - - - For tests not compatible with Azure Data Warehouse - 
-			xAzureSQLMI - - - - For tests not compatible with Azure SQL Managed Instance 
-			NTLM  - - - - - - - For tests using NTLM Authentication mode (excluded by default) 
-			reqExternalSetup  - For tests requiring external setup (excluded by default)
-			clientCertAuth  - - For tests requiring client certificate authentication setup (excluded by default)
-			- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - 
+			xJDBC42 - - - - - For tests not compatible with JDBC 42 Specifications - 
+			xGradle - - - - - For tests not compatible with Gradle Script - - - - - - 
+			xSQLv12 - - - - - For tests not compatible with SQL Server 2008 R2 - 2014 
+			xSQLv14 - - - - - For tests not compatible with SQL Server 2016 - 2017 - 
+			xSQLv15 - - - - - For tests not compatible with SQL Server 2019 - - - - - 
+			xAzureSQLDB - - - For tests not compatible with Azure SQL Database - - - 
+			xAzureSQLDW - - - For tests not compatible with Azure Data Warehouse - - 
+			xAzureSQLMI - - - For tests not compatible with Azure SQL Managed Instance 
+			NTLM - - - - - - For tests using NTLM Authentication mode (excluded by default) 
+			reqExternalSetup For tests requiring external setup (excluded by default) 
+			clientCertAuth - For tests requiring client certificate auth setup (excluded 
+			by default) - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - 
 			Default testing enabled with SQL Server 2019 (SQLv15) -->
-		<excludedGroups>xSQLv12,xSQLv15,NTLM,MSI,reqExternalSetup,clientCertAuth,fedAuth</excludedGroups>
 
-		<!-- Use -preview for preview release, leave empty for official release.-->
+		<!-- Use -preview for preview release, leave empty for official release. -->
 		<releaseExt>-preview</releaseExt>
 
 		<!-- Driver Dependencies -->
@@ -288,7 +287,8 @@
 						<configuration>
 							<!-- Exclude [xJDBC42] For tests not compatible with JDBC 4.2 Specifications -->
 							<excludedGroups>${excludedGroups}, xJDBC42</excludedGroups>
-							<!-- needs to be overridden as illegal-access and MaxPermSize=256m not valid in java 8 -->
+							<!-- needs to be overridden as illegal-access and MaxPermSize=256m 
+								not valid in java 8 -->
 							<argLine>-Xmx1024m -Djava.library.path=${dllPath}</argLine>
 						</configuration>
 					</plugin>
@@ -329,12 +329,12 @@
 			</build>
 		</profile>
 		<profile>
-			<id>jre16</id>
+			<id>jre17</id>
 			<activation>
 				<activeByDefault>true</activeByDefault>
 			</activation>
 			<build>
-				<finalName>${project.artifactId}-${project.version}.jre16${releaseExt}</finalName>
+				<finalName>${project.artifactId}-${project.version}.jre17${releaseExt}</finalName>
 				<plugins>
 					<plugin>
 						<groupId>org.apache.maven.plugins</groupId>
@@ -344,8 +344,8 @@
 							<excludes>
 								<exclude>**/com/microsoft/sqlserver/jdbc/SQLServerJdbc42.java</exclude>
 							</excludes>
-							<source>16</source>
-							<target>16</target>
+							<source>17</source>
+							<target>17</target>
 						</configuration>
 					</plugin>
 					<plugin>
@@ -429,8 +429,7 @@
 				<configuration>
 					<forkCount>3</forkCount>
 					<reuseForks>true</reuseForks>
-					<argLine>${argLine} --illegal-access=permit -Xmx1024m
-						-XX:MaxPermSize=256m -Djava.library.path=${dllPath}
+					<argLine>${argLine} -Xmx1024m -Djava.library.path=${dllPath}
 					</argLine>
 					<excludedGroups>${excludedGroups}</excludedGroups>
 				</configuration>
@@ -506,7 +505,7 @@
 			<plugin>
 				<groupId>org.jacoco</groupId>
 				<artifactId>jacoco-maven-plugin</artifactId>
-				<version>0.8.6</version>
+				<version>0.8.7</version>
 				<executions>
 					<execution>
 						<id>pre-test</id>
@@ -523,6 +522,10 @@
 					</execution>
 				</executions>
 				<configuration>
+					<excludes>
+						<exclude>**/mssql/googlecode/**/*</exclude>
+						<exclude>**/mssql/security/**/*</exclude>
+					</excludes>
 					<fileSets>
 						<fileSet>
 							<directory>jacoco-execs/</directory>
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/Column.java b/src/main/java/com/microsoft/sqlserver/jdbc/Column.java
index fd6b0e8f..ca913424 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/Column.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/Column.java
@@ -186,15 +186,15 @@ final class Column {
      * If the column has not yet been read from the response then this method reads it.
      */
     Object getValue(JDBCType jdbcType, InputStreamGetterArgs getterArgs, Calendar cal,
-            TDSReader tdsReader) throws SQLServerException {
+            TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException {
         Object value = getterDTV.getValue(jdbcType, typeInfo.getScale(), getterArgs, cal, typeInfo, cryptoMetadata,
-                tdsReader);
+                tdsReader, statement);
         setInternalVariant(getterDTV.getInternalVariant());
         return (null != filter) ? filter.apply(value, jdbcType) : value;
     }
 
-    int getInt(TDSReader tdsReader) throws SQLServerException {
-        return (Integer) getValue(JDBCType.INTEGER, null, null, tdsReader);
+    int getInt(TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException {
+        return (Integer) getValue(JDBCType.INTEGER, null, null, tdsReader, statement);
     }
 
     void updateValue(JDBCType jdbcType, Object value, JavaType javaType, StreamSetterArgs streamSetterArgs,
@@ -429,7 +429,7 @@ final class Column {
         updaterDTV = null;
     }
 
-    void sendByRPC(TDSWriter tdsWriter, SQLServerConnection conn) throws SQLServerException {
+    void sendByRPC(TDSWriter tdsWriter, SQLServerStatement statement) throws SQLServerException {
         // If the column has had no updates then there is nothing to send
         if (null == updaterDTV)
             return;
@@ -449,7 +449,7 @@ final class Column {
                                                                                                                        // for
                                                                                                                        // column
                                                                                                                        // updates)
-                    tdsWriter, conn);
+                    tdsWriter, statement);
         } finally {
             // this is for updateRow() stuff
             updaterDTV.sendCryptoMetaData(null, tdsWriter);
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java b/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java
index ab9e7226..0ce340e8 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java
@@ -1522,6 +1522,7 @@ final class TDSChannel implements Serializable {
             }
 
             boolean isServerNameValidated;
+            String dnsNameInSANCert = "";
 
             // the name in cert is in RFC2253 format parse it to get the actual subject name
             String subjectCN = parseCommonName(nameInCertDN);
@@ -1562,7 +1563,7 @@ final class TDSChannel implements Serializable {
                                 // Java docs guarantee that the second entry in the list
                                 // will be a string for dnsName
                                 if (value != null && value instanceof String) {
-                                    String dnsNameInSANCert = (String) value;
+                                    dnsNameInSANCert = (String) value;
 
                                     // Use English locale to avoid Turkish i issues.
                                     // Note that, this conversion was not necessary for
@@ -1599,8 +1600,9 @@ final class TDSChannel implements Serializable {
             }
 
             if (!isServerNameValidated) {
-                String msg = SQLServerException.getErrString("R_certNameFailed");
-                throw new CertificateException(msg);
+                MessageFormat form = new MessageFormat(SQLServerException.getErrString("R_certNameFailed"));
+                Object[] msgArgs = {hostName, dnsNameInSANCert};
+                throw new CertificateException(form.format(msgArgs));
             }
         }
 
@@ -4016,9 +4018,7 @@ final class TDSWriter {
      * efficiency. As this method will only be used in bulk copy, it needs to be efficient. Note: Any changes in
      * algorithm/logic should propagate to both writeReader() and writeNonUnicodeReader().
      */
-
-    void writeNonUnicodeReader(Reader reader, long advertisedLength, boolean isDestBinary,
-            Charset charSet) throws SQLServerException {
+    void writeNonUnicodeReader(Reader reader, long advertisedLength, boolean isDestBinary) throws SQLServerException {
         assert DataTypes.UNKNOWN_STREAM_LENGTH == advertisedLength || advertisedLength >= 0;
 
         long actualLength = 0;
@@ -4055,18 +4055,30 @@ final class TDSWriter {
                 // The Do-While loop goes on one more time as charsToWrite is greater than 0 for the last chunk, and
                 // in this last round the only thing that is written is an int value of 0, which is the PLP Terminator
                 // token(0x00000000).
-                writeInt(charsToWrite);
 
-                for (int charsCopied = 0; charsCopied < charsToWrite; ++charsCopied) {
-                    if (null == charSet) {
+                // collation from database is the collation used
+                Charset charSet = con.getDatabaseCollation().getCharset();
+
+                if (null == charSet) {
+                    writeInt(charsToWrite);
+
+                    for (int charsCopied = 0; charsCopied < charsToWrite; ++charsCopied) {
                         streamByteBuffer[charsCopied] = (byte) (streamCharBuffer[charsCopied] & 0xFF);
-                    } else {
-                        // encoding as per collation
-                        streamByteBuffer[charsCopied] = new String(streamCharBuffer[charsCopied] + "")
-                                .getBytes(charSet)[0];
                     }
+
+                    writeBytes(streamByteBuffer, 0, charsToWrite);
+                } else {
+                    bytesToWrite = 0;
+                    byte[] charBytes;
+                    for (int charsCopied = 0; charsCopied < charsToWrite; ++charsCopied) {
+                        charBytes = new String(streamCharBuffer[charsCopied] + "").getBytes(charSet);
+                        System.arraycopy(charBytes, 0, streamByteBuffer, bytesToWrite, charBytes.length);
+                        bytesToWrite += charBytes.length;
+                    }
+
+                    writeInt(bytesToWrite);
+                    writeBytes(streamByteBuffer, 0, bytesToWrite);
                 }
-                writeBytes(streamByteBuffer, 0, charsToWrite);
             } else {
                 bytesToWrite = charsToWrite;
                 if (0 != charsToWrite)
@@ -4980,7 +4992,7 @@ final class TDSWriter {
                         // Null header for v*max types is 0xFFFFFFFFFFFFFFFF.
                         writeLong(0xFFFFFFFFFFFFFFFFL);
                     } else if (isSqlVariant) {
-                        // for now we send as bigger type, but is sendStringParameterAsUnicoe is set to false we can't
+                        // for now we send as bigger type, but is sendStringParameterAsUnicode is set to false we can't
                         // send nvarchar
                         // since we are writing as nvarchar we need to write as tdstype.bigvarchar value because if we
                         // want to supprot varchar(8000) it becomes as nvarchar, 8000*2 therefore we should send as
@@ -5569,7 +5581,7 @@ final class TDSWriter {
     }
 
     void writeEncryptedRPCTime(String sName, GregorianCalendar localCalendar, int subSecondNanos, int scale,
-            boolean bOut) throws SQLServerException {
+            boolean bOut, SQLServerStatement statement) throws SQLServerException {
         if (con.getSendTimeAsDatetime()) {
             throw new SQLServerException(SQLServerException.getErrString("R_sendTimeAsDateTimeForAE"), null);
         }
@@ -5578,15 +5590,16 @@ final class TDSWriter {
         if (null == localCalendar)
             writeEncryptedRPCByteArray(null);
         else
-            writeEncryptedRPCByteArray(
-                    writeEncryptedScaledTemporal(localCalendar, subSecondNanos, scale, SSType.TIME, (short) 0));
+            writeEncryptedRPCByteArray(writeEncryptedScaledTemporal(localCalendar, subSecondNanos, scale, SSType.TIME,
+                    (short) 0, statement));
 
         writeByte(TDSType.TIMEN.byteValue());
         writeByte((byte) scale);
         writeCryptoMetaData();
     }
 
-    void writeEncryptedRPCDate(String sName, GregorianCalendar localCalendar, boolean bOut) throws SQLServerException {
+    void writeEncryptedRPCDate(String sName, GregorianCalendar localCalendar, boolean bOut,
+            SQLServerStatement statement) throws SQLServerException {
         writeRPCNameValType(sName, bOut, TDSType.BIGVARBINARY);
 
         if (null == localCalendar)
@@ -5595,14 +5608,14 @@ final class TDSWriter {
             writeEncryptedRPCByteArray(writeEncryptedScaledTemporal(localCalendar, 0, // subsecond nanos (none for a
                                                                                       // date value)
                     0, // scale (dates are not scaled)
-                    SSType.DATE, (short) 0));
+                    SSType.DATE, (short) 0, statement));
 
         writeByte(TDSType.DATEN.byteValue());
         writeCryptoMetaData();
     }
 
     void writeEncryptedRPCDateTime(String sName, GregorianCalendar cal, int subSecondNanos, boolean bOut,
-            JDBCType jdbcType) throws SQLServerException {
+            JDBCType jdbcType, SQLServerStatement statement) throws SQLServerException {
         assert (subSecondNanos >= 0) && (subSecondNanos < Nanos.PER_SECOND) : "Invalid subNanoSeconds value: "
                 + subSecondNanos;
         assert (cal != null) || (subSecondNanos == 0) : "Invalid subNanoSeconds value when calendar is null: "
@@ -5613,7 +5626,7 @@ final class TDSWriter {
         if (null == cal)
             writeEncryptedRPCByteArray(null);
         else
-            writeEncryptedRPCByteArray(getEncryptedDateTimeAsBytes(cal, subSecondNanos, jdbcType));
+            writeEncryptedRPCByteArray(getEncryptedDateTimeAsBytes(cal, subSecondNanos, jdbcType, statement));
 
         if (JDBCType.SMALLDATETIME == jdbcType) {
             writeByte(TDSType.DATETIMEN.byteValue());
@@ -5626,8 +5639,8 @@ final class TDSWriter {
     }
 
     // getEncryptedDateTimeAsBytes is called if jdbcType/ssType is SMALLDATETIME or DATETIME
-    byte[] getEncryptedDateTimeAsBytes(GregorianCalendar cal, int subSecondNanos,
-            JDBCType jdbcType) throws SQLServerException {
+    byte[] getEncryptedDateTimeAsBytes(GregorianCalendar cal, int subSecondNanos, JDBCType jdbcType,
+            SQLServerStatement statement) throws SQLServerException {
         int daysSinceSQLBaseDate = DDC.daysSinceBaseDate(cal.get(Calendar.YEAR), cal.get(Calendar.DAY_OF_YEAR),
                 TDS.BASE_YEAR_1900);
 
@@ -5681,7 +5694,7 @@ final class TDSWriter {
             byte[] value = new byte[4];
             System.arraycopy(days.array(), 0, value, 0, 2);
             System.arraycopy(seconds.array(), 0, value, 2, 2);
-            return SQLServerSecurityUtility.encryptWithKey(value, cryptoMeta, con);
+            return SQLServerSecurityUtility.encryptWithKey(value, cryptoMeta, con, statement);
         } else if (JDBCType.DATETIME == jdbcType) {
             // Last-ditch verification that the value is in the valid range for the
             // DATETIMEN TDS data type (1/1/1753 to 12/31/9999). If it's not, then
@@ -5707,7 +5720,7 @@ final class TDSWriter {
             byte[] value = new byte[8];
             System.arraycopy(days.array(), 0, value, 0, 4);
             System.arraycopy(seconds.array(), 0, value, 4, 4);
-            return SQLServerSecurityUtility.encryptWithKey(value, cryptoMeta, con);
+            return SQLServerSecurityUtility.encryptWithKey(value, cryptoMeta, con, statement);
         }
 
         assert false : "Unexpected JDBCType type " + jdbcType;
@@ -5715,14 +5728,14 @@ final class TDSWriter {
     }
 
     void writeEncryptedRPCDateTime2(String sName, GregorianCalendar localCalendar, int subSecondNanos, int scale,
-            boolean bOut) throws SQLServerException {
+            boolean bOut, SQLServerStatement statement) throws SQLServerException {
         writeRPCNameValType(sName, bOut, TDSType.BIGVARBINARY);
 
         if (null == localCalendar)
             writeEncryptedRPCByteArray(null);
         else
-            writeEncryptedRPCByteArray(
-                    writeEncryptedScaledTemporal(localCalendar, subSecondNanos, scale, SSType.DATETIME2, (short) 0));
+            writeEncryptedRPCByteArray(writeEncryptedScaledTemporal(localCalendar, subSecondNanos, scale,
+                    SSType.DATETIME2, (short) 0, statement));
 
         writeByte(TDSType.DATETIME2N.byteValue());
         writeByte((byte) (scale));
@@ -5730,7 +5743,7 @@ final class TDSWriter {
     }
 
     void writeEncryptedRPCDateTimeOffset(String sName, GregorianCalendar utcCalendar, int minutesOffset,
-            int subSecondNanos, int scale, boolean bOut) throws SQLServerException {
+            int subSecondNanos, int scale, boolean bOut, SQLServerStatement statement) throws SQLServerException {
         writeRPCNameValType(sName, bOut, TDSType.BIGVARBINARY);
 
         if (null == utcCalendar)
@@ -5738,7 +5751,7 @@ final class TDSWriter {
         else {
             assert 0 == utcCalendar.get(Calendar.ZONE_OFFSET);
             writeEncryptedRPCByteArray(writeEncryptedScaledTemporal(utcCalendar, subSecondNanos, scale,
-                    SSType.DATETIMEOFFSET, (short) minutesOffset));
+                    SSType.DATETIMEOFFSET, (short) minutesOffset, statement));
         }
 
         writeByte(TDSType.DATETIMEOFFSETN.byteValue());
@@ -5939,11 +5952,13 @@ final class TDSWriter {
      *        the SQL Server data type (DATE, TIME, DATETIME2, or DATETIMEOFFSET)
      * @param minutesOffset
      *        the offset value for DATETIMEOFFSET
+     * @param statement
+     *        the SQLServerStatement used for encryption
      * @throws SQLServerException
      *         if an I/O error occurs or if the value is not in the valid range
      */
     byte[] writeEncryptedScaledTemporal(GregorianCalendar cal, int subSecondNanos, int scale, SSType ssType,
-            short minutesOffset) throws SQLServerException {
+            short minutesOffset, SQLServerStatement statement) throws SQLServerException {
         assert con.isKatmaiOrLater();
 
         assert SSType.DATE == ssType || SSType.TIME == ssType || SSType.DATETIME2 == ssType
@@ -6025,7 +6040,7 @@ final class TDSWriter {
             byte[] encodedBytes = scaledNanosToEncodedBytes(scaledNanos, encodedLength);
 
             if (SSType.TIME == ssType) {
-                byte[] cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytes, cryptoMeta, con);
+                byte[] cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytes, cryptoMeta, con, statement);
                 return cipherText;
             } else if (SSType.DATETIME2 == ssType) {
                 // for DATETIME2 sends both date and time part together for encryption
@@ -6082,7 +6097,7 @@ final class TDSWriter {
 
             byte[] cipherText;
             if (SSType.DATE == ssType) {
-                cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytes, cryptoMeta, con);
+                cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytes, cryptoMeta, con, statement);
             } else if (SSType.DATETIME2 == ssType) {
                 // for Max value, does not round up, do casting instead.
                 if (3652058 == daysIntoCE) { // 9999-12-31
@@ -6102,7 +6117,8 @@ final class TDSWriter {
                 // Copy the 3 byte date value
                 System.arraycopy(encodedBytes, 0, encodedBytesForEncryption, (encodedBytesForEncryption.length - 3), 3);
 
-                cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytesForEncryption, cryptoMeta, con);
+                cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytesForEncryption, cryptoMeta, con,
+                        statement);
             } else {
                 // for Max value, does not round up, do casting instead.
                 if (3652058 == daysIntoCE) { // 9999-12-31
@@ -6128,7 +6144,8 @@ final class TDSWriter {
                                 .putShort(minutesOffset).array(),
                         0, encodedBytesForEncryption, (encodedBytesForEncryption.length - 2), 2);
 
-                cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytesForEncryption, cryptoMeta, con);
+                cipherText = SQLServerSecurityUtility.encryptWithKey(encodedBytesForEncryption, cryptoMeta, con,
+                        statement);
             }
             return cipherText;
         }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerEnclaveProvider.java b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerEnclaveProvider.java
index 7e60feb0..f863a950 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerEnclaveProvider.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerEnclaveProvider.java
@@ -168,6 +168,8 @@ interface ISQLServerEnclaveProvider {
      *        param names
      * @param connection
      *        connection
+     * @param sqlServerStatement
+     *        sqlServerStatement
      * @param stmt
      *        statement
      * @param rs
@@ -178,11 +180,17 @@ interface ISQLServerEnclaveProvider {
      *         if error
      */
     default void processSDPEv1(String userSql, String preparedTypeDefinitions, Parameter[] params,
-            ArrayList<String> parameterNames, SQLServerConnection connection, PreparedStatement stmt, ResultSet rs,
+            ArrayList<String> parameterNames, SQLServerConnection connection, SQLServerStatement sqlServerStatement, PreparedStatement stmt, ResultSet rs,
             ArrayList<byte[]> enclaveRequestedCEKs) throws SQLException {
         Map<Integer, CekTableEntry> cekList = new HashMap<>();
         CekTableEntry cekEntry = null;
         boolean isRequestedByEnclave = false;
+        SQLServerStatement statement = (SQLServerStatement) ((SQLServerPreparedStatement) stmt);
+
+        if (null != sqlServerStatement && sqlServerStatement.hasColumnEncryptionKeyStoreProvidersRegistered()) {
+            statement.registerColumnEncryptionKeyStoreProvidersOnStatement(sqlServerStatement.statementColumnEncryptionKeyStoreProviders);
+        }
+        
         while (rs.next()) {
             int currentOrdinal = rs.getInt(DescribeParameterEncryptionResultSet1.KeyOrdinal.value());
             if (!cekList.containsKey(currentOrdinal)) {
@@ -215,7 +223,7 @@ interface ISQLServerEnclaveProvider {
             if (isRequestedByEnclave) {
                 byte[] keySignature = rs.getBytes(DescribeParameterEncryptionResultSet1.EnclaveCMKSignature.value());
                 String serverName = connection.getTrustedServerNameAE();
-                SQLServerSecurityUtility.verifyColumnMasterKeyMetadata(connection, keyStoreName, keyPath, serverName,
+                SQLServerSecurityUtility.verifyColumnMasterKeyMetadata(connection, statement, keyStoreName, keyPath, serverName,
                         isRequestedByEnclave, keySignature);
 
                 // DBID(4) + MDVER(8) + KEYID(2) + CEK(32) = 46
@@ -223,8 +231,9 @@ interface ISQLServerEnclaveProvider {
                 aev2CekEntry.order(ByteOrder.LITTLE_ENDIAN).putInt(dbID);
                 aev2CekEntry.put(mdVer);
                 aev2CekEntry.putShort((short) keyID);
-                aev2CekEntry.put(connection.getColumnEncryptionKeyStoreProvider(keyStoreName)
-                        .decryptColumnEncryptionKey(keyPath, algo, encryptedKey));
+
+                SQLServerColumnEncryptionKeyStoreProvider provider = SQLServerSecurityUtility.getColumnEncryptionKeyStoreProvider(keyStoreName, connection, statement);
+                aev2CekEntry.put(provider.decryptColumnEncryptionKey(keyPath, algo, encryptedKey));
                 enclaveRequestedCEKs.add(aev2CekEntry.array());
             }
         }
@@ -257,7 +266,7 @@ interface ISQLServerEnclaveProvider {
                         encType.value,
                         (byte) rs.getInt(DescribeParameterEncryptionResultSet2.NormalizationRuleVersion.value()));
                 // Decrypt the symmetric key.(This will also validate and throw if needed).
-                SQLServerSecurityUtility.decryptSymmetricKey(params[paramIndex].cryptoMeta, connection);
+                SQLServerSecurityUtility.decryptSymmetricKey(params[paramIndex].cryptoMeta, connection, statement);
             } else {
                 if (params[paramIndex].getForceEncryption()) {
                     MessageFormat form = new MessageFormat(
@@ -284,6 +293,8 @@ interface ISQLServerEnclaveProvider {
      * 
      * @param connection
      *        connection
+     * @param statement
+     *        statement
      * @param userSql
      *        user sql
      * @param preparedTypeDefinitions
@@ -296,7 +307,7 @@ interface ISQLServerEnclaveProvider {
      * @throws SQLServerException
      *         when an error occurs.
      */
-    ArrayList<byte[]> createEnclaveSession(SQLServerConnection connection, String userSql,
+    ArrayList<byte[]> createEnclaveSession(SQLServerConnection connection, SQLServerStatement statement, String userSql,
             String preparedTypeDefinitions, Parameter[] params,
             ArrayList<String> parameterNames) throws SQLServerException;
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/NTLMAuthentication.java b/src/main/java/com/microsoft/sqlserver/jdbc/NTLMAuthentication.java
index fb9bb06c..47c700cf 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/NTLMAuthentication.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/NTLMAuthentication.java
@@ -617,8 +617,10 @@ final class NTLMAuthentication extends SSPIAuthentication {
     private static byte[] md4(final byte[] str) {
         MD4 md = new MD4();
         md.reset();
-        md.update(str);
-        return md.digest();
+        md.update(str, 0, str.length);
+        byte[] hash = new byte[md.getDigestSize()];
+        md.doFinal(hash, 0);
+        return hash;
     }
 
     /**
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/Parameter.java b/src/main/java/com/microsoft/sqlserver/jdbc/Parameter.java
index 0245c2e0..0c7a5815 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/Parameter.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/Parameter.java
@@ -389,22 +389,22 @@ final class Parameter {
     }
 
     Object getValue(JDBCType jdbcType, InputStreamGetterArgs getterArgs, Calendar cal,
-            TDSReader tdsReader) throws SQLServerException {
+            TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException {
         if (null == getterDTV)
             getterDTV = new DTV();
 
         deriveTypeInfo(tdsReader);
         // If the parameter is not encrypted or column encryption is turned off (either at connection or
         // statement level), cryptoMeta would be null.
-        return getterDTV.getValue(jdbcType, outScale, getterArgs, cal, typeInfo, cryptoMeta, tdsReader);
+        return getterDTV.getValue(jdbcType, outScale, getterArgs, cal, typeInfo, cryptoMeta, tdsReader, statement);
     }
 
     Object getSetterValue() {
         return setterDTV.getSetterValue();
     }
 
-    int getInt(TDSReader tdsReader) throws SQLServerException {
-        Integer value = (Integer) getValue(JDBCType.INTEGER, null, null, tdsReader);
+    int getInt(TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException {
+        Integer value = (Integer) getValue(JDBCType.INTEGER, null, null, tdsReader, statement);
         return null != value ? value : 0;
     }
 
@@ -1149,14 +1149,15 @@ final class Parameter {
         return typeDefinition;
     }
 
-    void sendByRPC(TDSWriter tdsWriter, SQLServerConnection conn) throws SQLServerException {
+    void sendByRPC(TDSWriter tdsWriter, SQLServerStatement statement) throws SQLServerException {
         assert null != inputDTV : "Parameter was neither set nor registered";
-
+        SQLServerConnection conn = statement.connection;
+        
         try {
             inputDTV.sendCryptoMetaData(this.cryptoMeta, tdsWriter);
             inputDTV.setJdbcTypeSetByUser(getJdbcTypeSetByUser(), getValueLength());
             inputDTV.sendByRPC(name, null, conn.getDatabaseCollation(), valueLength, isOutput() ? outScale : scale,
-                    isOutput(), tdsWriter, conn);
+                    isOutput(), tdsWriter, statement);
         } finally {
             // reset the cryptoMeta in IOBuffer
             inputDTV.sendCryptoMetaData(null, tdsWriter);
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerAASEnclaveProvider.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerAASEnclaveProvider.java
index bb651f76..5df1f76e 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerAASEnclaveProvider.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerAASEnclaveProvider.java
@@ -62,7 +62,7 @@ public class SQLServerAASEnclaveProvider implements ISQLServerEnclaveProvider {
     }
 
     @Override
-    public ArrayList<byte[]> createEnclaveSession(SQLServerConnection connection, String userSql,
+    public ArrayList<byte[]> createEnclaveSession(SQLServerConnection connection, SQLServerStatement statement, String userSql,
             String preparedTypeDefinitions, Parameter[] params,
             ArrayList<String> parameterNames) throws SQLServerException {
         // Check if the session exists in our cache
@@ -73,7 +73,7 @@ public class SQLServerAASEnclaveProvider implements ISQLServerEnclaveProvider {
             this.enclaveSession = entry.getEnclaveSession();
             this.aasParams = (AASAttestationParameters) entry.getBaseAttestationRequest();
         }
-        ArrayList<byte[]> b = describeParameterEncryption(connection, userSql, preparedTypeDefinitions, params,
+        ArrayList<byte[]> b = describeParameterEncryption(connection, statement, userSql, preparedTypeDefinitions, params,
                 parameterNames);
         if (connection.enclaveEstablished()) {
             return b;
@@ -116,7 +116,7 @@ public class SQLServerAASEnclaveProvider implements ISQLServerEnclaveProvider {
         }
     }
 
-    private ArrayList<byte[]> describeParameterEncryption(SQLServerConnection connection, String userSql,
+    private ArrayList<byte[]> describeParameterEncryption(SQLServerConnection connection, SQLServerStatement statement, String userSql,
             String preparedTypeDefinitions, Parameter[] params,
             ArrayList<String> parameterNames) throws SQLServerException {
         ArrayList<byte[]> enclaveRequestedCEKs = new ArrayList<>();
@@ -128,7 +128,7 @@ public class SQLServerAASEnclaveProvider implements ISQLServerEnclaveProvider {
                     // Should never happen.
                     return enclaveRequestedCEKs;
                 }
-                processSDPEv1(userSql, preparedTypeDefinitions, params, parameterNames, connection, stmt, rs,
+                processSDPEv1(userSql, preparedTypeDefinitions, params, parameterNames, connection, statement, stmt, rs,
                         enclaveRequestedCEKs);
                 // Process the third resultset.
                 if (connection.isAEv2() && stmt.getMoreResults()) {
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCopy.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCopy.java
index a8405af0..558fb586 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCopy.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCopy.java
@@ -2263,22 +2263,22 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                                 } else {
                                     reader = new StringReader(colValue.toString());
                                 }
+
                                 if (unicodeConversionRequired(bulkJdbcType, destSSType)) {
                                     // writeReader is unicode.
                                     tdsWriter.writeReader(reader, DataTypes.UNKNOWN_STREAM_LENGTH, true);
                                 } else {
                                     if ((SSType.BINARY == destSSType) || (SSType.VARBINARY == destSSType)
                                             || (SSType.VARBINARYMAX == destSSType) || (SSType.IMAGE == destSSType)) {
-                                        tdsWriter.writeNonUnicodeReader(reader, DataTypes.UNKNOWN_STREAM_LENGTH, true,
-                                                null);
+                                        tdsWriter.writeNonUnicodeReader(reader, DataTypes.UNKNOWN_STREAM_LENGTH, true);
                                     } else {
                                         SQLCollation destCollation = destColumnMetadata.get(destColOrdinal).collation;
                                         if (null != destCollation) {
                                             tdsWriter.writeNonUnicodeReader(reader, DataTypes.UNKNOWN_STREAM_LENGTH,
-                                                    false, destCollation.getCharset());
+                                                    false);
                                         } else {
                                             tdsWriter.writeNonUnicodeReader(reader, DataTypes.UNKNOWN_STREAM_LENGTH,
-                                                    false, null);
+                                                    false);
                                         }
                                     }
                                 }
@@ -2292,21 +2292,23 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                         if (null == colValue) {
                             writeNullToTdsWriter(tdsWriter, bulkJdbcType, isStreaming);
                         } else {
-                            String colValueStr = colValue.toString();
 
-                            // Remove extra trailing zeros added from toString
-                            if (colValue instanceof LocalDateTime || colValue instanceof LocalTime) {
-                                colValueStr = colValueStr.contains(".") ? colValueStr.replaceAll("0*$", "")
-                                                                        : colValueStr;
+                            String colValueStr;
+                            if (colValue instanceof LocalDateTime) {
+                                colValueStr = ((LocalDateTime)colValue).format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
+                            } else if (colValue instanceof LocalTime) {
+                                colValueStr = ((LocalTime)colValue).format(DateTimeFormatter.ISO_LOCAL_TIME);
+                            } else {
+                                colValueStr = colValue.toString();
                             }
 
                             if (unicodeConversionRequired(bulkJdbcType, destSSType)) {
-                                int stringLength = colValue.toString().length();
+                                int stringLength = colValueStr.length();
                                 byte[] typevarlen = new byte[2];
                                 typevarlen[0] = (byte) (2 * stringLength & 0xFF);
                                 typevarlen[1] = (byte) ((2 * stringLength >> 8) & 0xFF);
                                 tdsWriter.writeBytes(typevarlen);
-                                tdsWriter.writeString(colValue.toString());
+                                tdsWriter.writeString(colValueStr);
                             } else {
                                 if ((SSType.BINARY == destSSType) || (SSType.VARBINARY == destSSType)) {
                                     byte[] bytes = null;
@@ -3092,7 +3094,7 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                 // normalize the values before encrypting them
                 colValue = SQLServerSecurityUtility.encryptWithKey(normalizedValue(destJdbcType, colValue,
                         baseSrcJdbcType, destTypeInfo.getPrecision(), destTypeInfo.getScale()), destCryptoMeta,
-                        connection);
+                        connection, null);
             }
         }
         writeColumnToTdsWriter(tdsWriter, srcPrecision, srcScale, srcJdbcType, srcNullable, srcColOrdinal,
@@ -3351,7 +3353,7 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                 calendar.setTimeInMillis(((Date) colValue).getTime());
                 return tdsWriter.writeEncryptedScaledTemporal(calendar, 0, // subsecond nanos (none for a date value)
                         0, // scale (dates are not scaled)
-                        SSType.DATE, (short) 0);
+                        SSType.DATE, (short) 0, null);
 
             case TIME:
                 calendar = new GregorianCalendar(java.util.TimeZone.getDefault(), java.util.Locale.US);
@@ -3367,7 +3369,8 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                     if (subSecondNanos < 0)
                         subSecondNanos += Nanos.PER_SECOND;
                 }
-                return tdsWriter.writeEncryptedScaledTemporal(calendar, subSecondNanos, scale, SSType.TIME, (short) 0);
+                return tdsWriter.writeEncryptedScaledTemporal(calendar, subSecondNanos, scale, SSType.TIME, (short) 0,
+                        null);
 
             case TIMESTAMP:
                 calendar = new GregorianCalendar(java.util.TimeZone.getDefault(), java.util.Locale.US);
@@ -3377,7 +3380,7 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                 calendar.setTimeInMillis(utcMillis);
                 subSecondNanos = ((java.sql.Timestamp) colValue).getNanos();
                 return tdsWriter.writeEncryptedScaledTemporal(calendar, subSecondNanos, scale, SSType.DATETIME2,
-                        (short) 0);
+                        (short) 0, null);
 
             case DATETIME:
             case SMALLDATETIME:
@@ -3387,7 +3390,7 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                 utcMillis = ((java.sql.Timestamp) colValue).getTime();
                 calendar.setTimeInMillis(utcMillis);
                 subSecondNanos = ((java.sql.Timestamp) colValue).getNanos();
-                return tdsWriter.getEncryptedDateTimeAsBytes(calendar, subSecondNanos, srcTemporalJdbcType);
+                return tdsWriter.getEncryptedDateTimeAsBytes(calendar, subSecondNanos, srcTemporalJdbcType, null);
 
             case DATETIMEOFFSET:
                 microsoft.sql.DateTimeOffset dtoValue = (microsoft.sql.DateTimeOffset) colValue;
@@ -3399,7 +3402,7 @@ public class SQLServerBulkCopy implements java.lang.AutoCloseable, java.io.Seria
                 calendar.clear();
                 calendar.setTimeInMillis(utcMillis);
                 return tdsWriter.writeEncryptedScaledTemporal(calendar, subSecondNanos, scale, SSType.DATETIMEOFFSET,
-                        (short) minutesOffset);
+                        (short) minutesOffset, null);
 
             default:
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerCallableStatement.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerCallableStatement.java
index 0a5439bc..095a526d 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerCallableStatement.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerCallableStatement.java
@@ -436,11 +436,11 @@ public class SQLServerCallableStatement extends SQLServerPreparedStatement imple
     }
 
     private Object getValue(int parameterIndex, JDBCType jdbcType) throws SQLServerException {
-        return getterGetParam(parameterIndex).getValue(jdbcType, null, null, resultsReader());
+        return getterGetParam(parameterIndex).getValue(jdbcType, null, null, resultsReader(), this);
     }
 
     private Object getValue(int parameterIndex, JDBCType jdbcType, Calendar cal) throws SQLServerException {
-        return getterGetParam(parameterIndex).getValue(jdbcType, null, cal, resultsReader());
+        return getterGetParam(parameterIndex).getValue(jdbcType, null, cal, resultsReader(), this);
     }
 
     private Object getStream(int parameterIndex, StreamType streamType) throws SQLServerException {
@@ -448,7 +448,8 @@ public class SQLServerCallableStatement extends SQLServerPreparedStatement imple
                 new InputStreamGetterArgs(streamType, getIsResponseBufferingAdaptive(),
                         getIsResponseBufferingAdaptive(), toString()),
                 null, // calendar
-                resultsReader());
+                resultsReader(),
+                this);
 
         activeStream = (Closeable) value;
         return value;
@@ -459,7 +460,8 @@ public class SQLServerCallableStatement extends SQLServerPreparedStatement imple
                 new InputStreamGetterArgs(StreamType.SQLXML, getIsResponseBufferingAdaptive(),
                         getIsResponseBufferingAdaptive(), toString()),
                 null, // calendar
-                resultsReader());
+                resultsReader(),
+                this);
 
         if (null != value)
             activeStream = value.getStream();
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java
index 3b31812a..80ccba0a 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java
@@ -17,6 +17,7 @@ import java.nio.ByteOrder;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.text.MessageFormat;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
@@ -41,6 +42,63 @@ import com.azure.security.keyvault.keys.models.KeyType;
 import com.azure.security.keyvault.keys.models.KeyVaultKey;
 
 
+class CMKMetadataSignatureInfo {
+    String masterKeyPath;
+    boolean allowEnclaveComputations;
+    String signatureHexString;
+
+    public CMKMetadataSignatureInfo(String masterKeyPath, boolean allowEnclaveComputations, byte[] signature) {
+        this.masterKeyPath = masterKeyPath;
+        this.allowEnclaveComputations = allowEnclaveComputations;
+        this.signatureHexString = Util.byteToHexDisplayString(signature);
+    }
+
+    public String getMasterKeyPath() {
+        return masterKeyPath;
+    }
+
+    public boolean isAllowEnclaveComputations() {
+        return allowEnclaveComputations;
+    }
+
+    public String getSignatureHexString() {
+        return signatureHexString;
+    }
+
+    @Override
+    public int hashCode() {
+        int hash = 7;
+        hash = 31 * hash + (null != masterKeyPath ? masterKeyPath.hashCode() : 0);
+        hash = 31 * hash + (allowEnclaveComputations ? 1 : 0);
+        hash = 31 * hash + (null != signatureHexString ? signatureHexString.hashCode() : 0);
+
+        return hash;
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+
+        if (null != object && CMKMetadataSignatureInfo.class == object.getClass()) {
+            CMKMetadataSignatureInfo other = (CMKMetadataSignatureInfo) object;
+
+            if (hashCode() == other.hashCode()) {
+                return ((null == masterKeyPath ? null == other.masterKeyPath
+                                               : masterKeyPath.equals(other.masterKeyPath))
+                        && allowEnclaveComputations == other.allowEnclaveComputations
+                        && (null == signatureHexString ? null == other.signatureHexString
+                                                       : signatureHexString.equals(other.signatureHexString)));
+            }
+
+        }
+
+        return false;
+    }
+}
+
+
 /**
  * Provides implementation similar to certificate store provider. A CEK encrypted with certificate store provider should
  * be decryptable by this provider and vice versa.
@@ -83,6 +141,19 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
     private Map<String, CryptographyClient> cachedCryptographyClients = new ConcurrentHashMap<>();
     private TokenCredential credential;
 
+    /**
+     * A cache of column encryption keys (once they are unwrapped). This is useful for rapidly decrypting multiple data
+     * values. The default expiration is set to 2 hours.
+     */
+    private final SimpleTtlCache<String, byte[]> columnEncryptionKeyCache = new SimpleTtlCache<>();
+
+    /**
+     * A cache for storing the results of signature verification of column master key metadata. The default expiration
+     * is set to 10 days.
+     */
+    private final SimpleTtlCache<CMKMetadataSignatureInfo, Boolean> cmkMetadataSignatureVerificationCache = new SimpleTtlCache<>(
+            Duration.ofDays(10));
+
     public void setName(String name) {
         this.name = name;
     }
@@ -91,6 +162,27 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
         return this.name;
     }
 
+    /**
+     * Returns the time-to-live for items in the columnEncryptionKeyCache.
+     * 
+     * @return the time-to-live for items in the columnEncryptionKeyCache.
+     */
+    @Override
+    public Duration getColumnEncryptionKeyCacheTtl() {
+        return columnEncryptionKeyCache.getCacheTtl();
+    }
+
+    /**
+     * Sets the the time-to-live for items in the columnEncryptionKeyCache.
+     * 
+     * @param duration
+     *        value to be set for the time-to-live for items in the columnEncryptionKeyCache.
+     */
+    @Override
+    public void setColumnEncryptionCacheTtl(Duration duration) {
+        columnEncryptionKeyCache.setCacheTtl(duration);
+    }
+
     /**
      * Constructs a SQLServerColumnEncryptionAzureKeyVaultProvider to authenticate to AAD using the client id and client
      * key. This is used by KeyVault client at runtime to authenticate to Azure Key Vault.
@@ -261,6 +353,17 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
             throw new SQLServerException(this, form.format(msgArgs), null, 0, false);
         }
 
+        // check cache if TimeToLive is not 0
+        boolean allowCache = false;
+        String encryptedColumnEncryptionKeyHexString = Util.byteToHexDisplayString(encryptedColumnEncryptionKey);
+        if (columnEncryptionKeyCache.getCacheTtl().getSeconds() > 0) {
+            allowCache = true;
+
+            if (columnEncryptionKeyCache.contains(encryptedColumnEncryptionKeyHexString)) {
+                return columnEncryptionKeyCache.get(encryptedColumnEncryptionKeyHexString);
+            }
+        }
+
         // Get key path length
         int currentIndex = firstVersion.length;
         short keyPathLength = convertTwoBytesToShort(encryptedColumnEncryptionKey, currentIndex);
@@ -329,6 +432,10 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
         // Decrypt the CEK
         byte[] decryptedCEK = this.AzureKeyVaultUnWrap(masterKeyPath, keyWrapAlgorithm, cipherText);
 
+        if (allowCache) {
+            columnEncryptionKeyCache.put(encryptedColumnEncryptionKeyHexString, decryptedCEK);
+        }
+
         return decryptedCEK;
     }
 
@@ -759,6 +866,12 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
 
         KeyStoreProviderCommon.validateNonEmptyMasterKeyPath(masterKeyPath);
 
+        CMKMetadataSignatureInfo key = new CMKMetadataSignatureInfo(masterKeyPath, allowEnclaveComputations, signature);
+
+        if (cmkMetadataSignatureVerificationCache.contains(key)) {
+            return cmkMetadataSignatureVerificationCache.get(key);
+        }
+
         try {
             MessageDigest md = MessageDigest.getInstance("SHA-256");
             md.update(name.toLowerCase().getBytes(java.nio.charset.StandardCharsets.UTF_16LE));
@@ -778,7 +891,10 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
             }
 
             // Validate the signature
-            return AzureKeyVaultVerifySignature(dataToVerify, signature, masterKeyPath);
+            boolean isValid = AzureKeyVaultVerifySignature(dataToVerify, signature, masterKeyPath);
+            cmkMetadataSignatureVerificationCache.put(key, isValid);
+
+            return isValid;
         } catch (NoSuchAlgorithmException e) {
             throw new SQLServerException(SQLServerException.getErrString("R_NoSHA256Algorithm"), e);
         }
@@ -840,4 +956,12 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
         }
         return (null != props && !props.isEmpty()) ? props : null;
     }
+
+    int getColumnEncryptionKeyCacheSize() {
+        return columnEncryptionKeyCache.getCacheSize();
+    }
+
+    int getCmkMetadataSignatureVerificationCacheSize() {
+        return cmkMetadataSignatureVerificationCache.getCacheSize();
+    }
 }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionKeyStoreProvider.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionKeyStoreProvider.java
index 6a11025a..93d4cce5 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionKeyStoreProvider.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionKeyStoreProvider.java
@@ -5,9 +5,11 @@
 
 package com.microsoft.sqlserver.jdbc;
 
+import java.time.Duration;
+
 /**
  * 
- * Defines the abtract class for a SQL Server Column Encryption key store provider Extend this class to implement a
+ * Defines the abstract class for a SQL Server Column Encryption key store provider Extend this class to implement a
  * custom key store provider.
  *
  */
@@ -77,4 +79,22 @@ public abstract class SQLServerColumnEncryptionKeyStoreProvider {
      *      when an error occurs while verifying the signature
      */
     public abstract boolean verifyColumnMasterKeyMetadata (String masterKeyPath, boolean allowEnclaveComputations, byte[] signature) throws SQLServerException;
+
+    /**
+     * Returns the time-to-live for items in the cache of column encryption keys, as implemented in the key store provider.
+     * 
+     * @return the time-to-live for items in the cache.
+     */
+    public Duration getColumnEncryptionKeyCacheTtl() {
+        return Duration.ZERO;
+    }
+
+    /**
+     * Sets the the time-to-live for items in the cache of column encryption keys in the key store provider.
+     * 
+     * @param duration
+     *        value to be set for the time-to-live for items in the cache in the key store provider.
+     */
+    public void setColumnEncryptionCacheTtl(Duration duration) {
+    }
 }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java
index ef7b3a83..3f5730c7 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java
@@ -28,6 +28,7 @@ import java.sql.SQLXML;
 import java.sql.Savepoint;
 import java.sql.Statement;
 import java.text.MessageFormat;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Enumeration;
@@ -746,6 +747,8 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
     static final String RESERVED_PROVIDER_NAME_PREFIX = "MSSQL_";
 
+    static final String WINDOWS_KEY_STORE_NAME = "MSSQL_CERTIFICATE_STORE";
+
     /** column encryption setting */
     String columnEncryptionSetting = null;
 
@@ -826,13 +829,16 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
     /** global system ColumnEncryptionKeyStoreProviders */
     static Map<String, SQLServerColumnEncryptionKeyStoreProvider> globalSystemColumnEncryptionKeyStoreProviders = new HashMap<>();
+
     static {
         if (System.getProperty("os.name").toLowerCase(Locale.ENGLISH).startsWith("windows")) {
             SQLServerColumnEncryptionCertificateStoreProvider provider = new SQLServerColumnEncryptionCertificateStoreProvider();
             globalSystemColumnEncryptionKeyStoreProviders.put(provider.getName(), provider);
         }
     }
+
     static Map<String, SQLServerColumnEncryptionKeyStoreProvider> globalCustomColumnEncryptionKeyStoreProviders = null;
+
     /** This is a per-connection store provider. It can be JKS or AKV. */
     Map<String, SQLServerColumnEncryptionKeyStoreProvider> systemColumnEncryptionKeyStoreProvider = new HashMap<>();
 
@@ -864,7 +870,7 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
         for (Map.Entry<String, SQLServerColumnEncryptionKeyStoreProvider> entry : clientKeyStoreProviders.entrySet()) {
             String providerName = entry.getKey();
-            if (null == providerName || 0 == providerName.length()) {
+            if (null == providerName || 0 == providerName.trim().length()) {
                 throw new SQLServerException(null, SQLServerException.getErrString("R_EmptyCustomKeyStoreProviderName"),
                         null, 0, false);
             }
@@ -874,13 +880,17 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
                 Object[] msgArgs = {providerName, RESERVED_PROVIDER_NAME_PREFIX};
                 throw new SQLServerException(null, form.format(msgArgs), null, 0, false);
             }
-            if (null == entry.getValue()) {
-                MessageFormat form = new MessageFormat(
-                        SQLServerException.getErrString("R_CustomKeyStoreProviderValueNull"));
-                Object[] msgArgs = {providerName, RESERVED_PROVIDER_NAME_PREFIX};
-                throw new SQLServerException(null, form.format(msgArgs), null, 0, false);
+
+            SQLServerColumnEncryptionKeyStoreProvider provider = entry.getValue();
+            if (null == provider) {
+                throw new SQLServerException(null,
+                        String.format(SQLServerException.getErrString("R_CustomKeyStoreProviderValueNull"), providerName),
+                        null, 0, false);
             }
-            globalCustomColumnEncryptionKeyStoreProviders.put(entry.getKey(), entry.getValue());
+
+            // Global providers should not use their own CEK caches.
+            provider.setColumnEncryptionCacheTtl(Duration.ZERO);
+            globalCustomColumnEncryptionKeyStoreProviders.put(providerName, provider);
         }
 
         loggerExternal.exiting(loggingClassNameBase, "registerColumnEncryptionKeyStoreProviders",
@@ -937,15 +947,16 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
                 .containsKey(providerName)) ? systemColumnEncryptionKeyStoreProvider.get(providerName) : null;
     }
 
-    synchronized SQLServerColumnEncryptionKeyStoreProvider getColumnEncryptionKeyStoreProvider(
+    synchronized SQLServerColumnEncryptionKeyStoreProvider getSystemOrGlobalColumnEncryptionKeyStoreProvider(
             String providerName) throws SQLServerException {
+        
+        // check for global system providers
+        keystoreProvider = getGlobalSystemColumnEncryptionKeyStoreProvider(providerName);
 
-        // Check for the connection provider first.
-        keystoreProvider = getSystemColumnEncryptionKeyStoreProvider(providerName);
-
-        // There is no connection provider of this name, check for the global system providers.
-        if (null == keystoreProvider) {
-            keystoreProvider = getGlobalSystemColumnEncryptionKeyStoreProvider(providerName);
+        // Check for the connection system provider.
+        if (null == keystoreProvider)
+        {
+            keystoreProvider = getSystemColumnEncryptionKeyStoreProvider(providerName);
         }
 
         // There is no global system provider of this name, check for the global custom providers.
@@ -966,6 +977,90 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
         return keystoreProvider;
     }
 
+    synchronized boolean hasConnectionColumnEncryptionKeyStoreProvidersRegistered() {
+        return null != connectionColumnEncryptionKeyStoreProvider && connectionColumnEncryptionKeyStoreProvider.size() > 0;
+    }
+
+    synchronized String getAllConnectionColumnEncryptionKeyStoreProviders() {
+        String keyStores = "";
+        if (0 != connectionColumnEncryptionKeyStoreProvider.size())
+            keyStores = connectionColumnEncryptionKeyStoreProvider.keySet().toString();
+        return keyStores;
+    }
+
+    synchronized SQLServerColumnEncryptionKeyStoreProvider getColumnEncryptionKeyStoreProviderOnConnection(
+            String providerName) throws SQLServerException {
+
+        // Check for a connection-level provider first
+        if (null != connectionColumnEncryptionKeyStoreProvider &&
+                connectionColumnEncryptionKeyStoreProvider.size() > 0) {
+            // If any connection-level providers are registered, we don't fall back to system/global providers
+            if (connectionColumnEncryptionKeyStoreProvider.containsKey(providerName)) {
+                return connectionColumnEncryptionKeyStoreProvider.get(providerName);
+            } else {
+                MessageFormat form = new MessageFormat(
+                        SQLServerException.getErrString("R_UnrecognizedConnectionKeyStoreProviderName"));
+                Object[] msgArgs = {providerName, getAllConnectionColumnEncryptionKeyStoreProviders()};
+                throw new SQLServerException(form.format(msgArgs), null);
+            }
+        }
+
+        // No connection-level providers registered, so return system/global provider
+        return getSystemOrGlobalColumnEncryptionKeyStoreProvider(providerName);
+    }
+
+    /** This is a user-defined per-connection store provider. */
+    Map<String, SQLServerColumnEncryptionKeyStoreProvider> connectionColumnEncryptionKeyStoreProvider = new HashMap<>();
+
+    /**
+     * Registers connection-level key store providers, replacing all existing providers.
+     * 
+     * @param clientKeyStoreProviders
+     *        a map containing the store providers information.
+     * @throws SQLServerException
+     *         when an error occurs
+     */
+    public synchronized void registerColumnEncryptionKeyStoreProvidersOnConnection(
+            Map<String, SQLServerColumnEncryptionKeyStoreProvider> clientKeyStoreProviders) throws SQLServerException {
+        loggerExternal.entering(loggingClassName, "registerColumnEncryptionKeyStoreProvidersOnConnection",
+                "Registering Column Encryption Key Store Providers on Connection");
+
+        if (null == clientKeyStoreProviders) {
+            throw new SQLServerException(null, SQLServerException.getErrString("R_CustomKeyStoreProviderMapNull"), null,
+                    0, false);
+        }
+
+        connectionColumnEncryptionKeyStoreProvider.clear();
+
+        for (Map.Entry<String, SQLServerColumnEncryptionKeyStoreProvider> entry : clientKeyStoreProviders.entrySet()) {
+            String providerName = entry.getKey();
+            if (null == providerName || 0 == providerName.trim().length()) {
+                throw new SQLServerException(null, SQLServerException.getErrString("R_EmptyCustomKeyStoreProviderName"),
+                        null, 0, false);
+            }
+
+            // MSSQL_CERTIFICATE_STORE not allowed on connection level
+            if ((providerName.equalsIgnoreCase(WINDOWS_KEY_STORE_NAME))) {
+                MessageFormat form = new MessageFormat(
+                        SQLServerException.getErrString("R_InvalidCustomKeyStoreProviderName"));
+                Object[] msgArgs = {providerName, WINDOWS_KEY_STORE_NAME};
+                throw new SQLServerException(null, form.format(msgArgs), null, 0, false);
+            }
+
+            if (null == entry.getValue()) {
+                throw new SQLServerException(null,
+                        String.format(SQLServerException.getErrString("R_CustomKeyStoreProviderValueNull"), providerName),
+                        null, 0, false);
+            }            
+
+            connectionColumnEncryptionKeyStoreProvider.put(entry.getKey(), entry.getValue());
+        }
+
+        loggerExternal.exiting(loggingClassName, "registerColumnEncryptionKeyStoreProvidersOnConnection",
+                "Number of connection-level Key store providers that are registered: "
+                        + connectionColumnEncryptionKeyStoreProvider.size());
+    }
+
     /** trusted servername AE */
     private String trustedServerNameAE = null;
     private static Map<String, List<String>> columnEncryptionTrustedMasterKeyPaths = new HashMap<>();
@@ -1215,6 +1310,8 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
     static final private java.util.logging.Logger loggerExternal = java.util.logging.Logger
             .getLogger("com.microsoft.sqlserver.jdbc.Connection");
     private static String loggingClassNameBase = "com.microsoft.sqlserver.jdbc.SQLServerConnection";
+
+    /** Instance-specific loggingClassName to identity the connection in logs */
     private String loggingClassName = loggingClassNameBase;
 
     /**
@@ -6531,12 +6628,13 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
     /**
      * Sets time-to-live for column encryption key entries in the column encryption key cache for the Always Encrypted
-     * feature. The default value is 2 hours. This variable holds the value in seconds.
+     * feature. The default value is 2 hours. This variable holds the value in seconds. This only applies to global-level
+     * key store providers. Connection and Statement-level providers need to set their own cache TTL values.
      * 
      * @param columnEncryptionKeyCacheTTL
      *        The timeunit in seconds
      * @param unit
-     *        The Timeunit.
+     *        The Timeunit
      * @throws SQLServerException
      *         when an error occurs
      */
@@ -6909,12 +7007,12 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
     /** Enclave provider */
     private ISQLServerEnclaveProvider enclaveProvider;
 
-    ArrayList<byte[]> initEnclaveParameters(String userSql, String preparedTypeDefinitions, Parameter[] params,
+    ArrayList<byte[]> initEnclaveParameters(SQLServerStatement statement, String userSql, String preparedTypeDefinitions, Parameter[] params,
             ArrayList<String> parameterNames) throws SQLServerException {
         if (!this.enclaveEstablished()) {
             enclaveProvider.getAttestationParameters(this.enclaveAttestationUrl);
         }
-        return enclaveProvider.createEnclaveSession(this, userSql, preparedTypeDefinitions, params, parameterNames);
+        return enclaveProvider.createEnclaveSession(this, statement, userSql, preparedTypeDefinitions, params, parameterNames);
     }
 
     boolean enclaveEstablished() {
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java
index f053a828..764deb8d 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java
@@ -62,7 +62,7 @@ public final class SQLServerError extends StreamPacket implements Serializable {
          */
         SQLSERVER_ERROR_4221(4221),
 
-        // The follow are from SqlClient
+        // The following are from SqlClient
 
         // Resource ID: %d. The %s limit for the database is %d and has been reached.
         SQLSERVER_ERROR_10928(10928),
@@ -72,7 +72,13 @@ public final class SQLServerError extends StreamPacket implements Serializable {
          * Resource ID: %d. The %s minimum guarantee is %d, maximum limit is %d and the current usage for the database
          * is %d. However, the server is currently too busy to support requests greater than %d for this database.
          */
-        SQLSERVER_ERROR_10929(10929);
+        SQLSERVER_ERROR_10929(10929),
+
+        // Can not connect to the SQL pool since it is paused. Please resume the SQL pool and try again.
+        SQLSERVER_ERROR_42108(42108),
+
+        // The SQL pool is warming up. Please try again.
+        SQLSERVER_ERROR_42109(42109);
 
         private final int errNo;
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerLexer.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerLexer.java
index 71a6b64b..3809f3c8 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerLexer.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerLexer.java
@@ -20,7 +20,7 @@ import org.antlr.v4.runtime.dfa.DFA;
 @SuppressWarnings({"all", "warnings", "unchecked", "unused", "cast"})
 class SQLServerLexer extends Lexer {
     static {
-        RuntimeMetaData.checkVersion("4.7.2", RuntimeMetaData.VERSION);
+        RuntimeMetaData.checkVersion("4.9.2", RuntimeMetaData.VERSION);
     }
 
     protected static final DFA[] _decisionToDFA;
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java
index ab46fc0b..8273b4b9 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java
@@ -570,7 +570,7 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
         }
 
         if (connection.isAEv2() && !isInternalEncryptionQuery) {
-            this.enclaveCEKs = connection.initEnclaveParameters(preparedSQL, preparedTypeDefinitions, inOutParam,
+            this.enclaveCEKs = connection.initEnclaveParameters(this, preparedSQL, preparedTypeDefinitions, inOutParam,
                     parameterNames);
             encryptionMetadataIsRetrieved = true;
             setMaxRowsAndMaxFieldSize();
@@ -656,6 +656,11 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
      */
     boolean consumeExecOutParam(TDSReader tdsReader) throws SQLServerException {
         final class PrepStmtExecOutParamHandler extends StmtExecOutParamHandler {
+
+            PrepStmtExecOutParamHandler(SQLServerStatement statement) {
+                super(statement);
+            }
+
             boolean onRetValue(TDSReader tdsReader) throws SQLServerException {
                 // If no prepared statement handle is expected at this time
                 // then don't consume this OUT parameter as it does not contain
@@ -670,7 +675,7 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
                         Util.shouldHonorAEForParameters(stmtColumnEncriptionSetting, connection));
                 param.skipRetValStatus(tdsReader);
 
-                setPreparedStatementHandle(param.getInt(tdsReader));
+                setPreparedStatementHandle(param.getInt(tdsReader, statement));
 
                 // Cache the reference to the newly created handle, NOT for cursorable handles.
                 if (null == cachedPreparedStatementHandle && !isCursorable(executeMethod)) {
@@ -688,7 +693,7 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
         }
 
         if (expectPrepStmtHandle || expectCursorOutParams) {
-            TDSParser.parse(tdsReader, new PrepStmtExecOutParamHandler());
+            TDSParser.parse(tdsReader, new PrepStmtExecOutParamHandler(this));
             return true;
         }
 
@@ -707,7 +712,7 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
                 tdsWriter.writeByte((byte) paramNameLen);
                 tdsWriter.writeString(new String(cParamName, 0, paramNameLen));
             }
-            params[index].sendByRPC(tdsWriter, connection);
+            params[index].sendByRPC(tdsWriter, this);
         }
     }
 
@@ -877,6 +882,12 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
                 getStatementLogger().fine(
                         "Calling stored procedure sp_describe_parameter_encryption to get parameter encryption information.");
             }
+
+            // pass registered custom provider to stmt
+            if (this.hasColumnEncryptionKeyStoreProvidersRegistered()) {
+                ((SQLServerCallableStatement) stmt).registerColumnEncryptionKeyStoreProvidersOnStatement(this.statementColumnEncryptionKeyStoreProviders);
+            }
+
             ((SQLServerCallableStatement) stmt).isInternalEncryptionQuery = true;
             ((SQLServerCallableStatement) stmt).setNString(1, preparedSQL);
             ((SQLServerCallableStatement) stmt).setNString(2, preparedTypeDefinitions);
@@ -943,8 +954,10 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
                                             DescribeParameterEncryptionResultSet2.ColumnEncryptionAlgorithm.value()),
                                     null, encType.value, (byte) secondRs.getInt(
                                             DescribeParameterEncryptionResultSet2.NormalizationRuleVersion.value()));
+
+                            SQLServerStatement statement = (SQLServerStatement) stmt;
                             // Decrypt the symmetric key.(This will also validate and throw if needed).
-                            SQLServerSecurityUtility.decryptSymmetricKey(params[paramIndex].cryptoMeta, connection);
+                            SQLServerSecurityUtility.decryptSymmetricKey(params[paramIndex].cryptoMeta, connection, statement);
                         } else {
                             if (params[paramIndex].getForceEncryption()) {
                                 MessageFormat form = new MessageFormat(SQLServerException
@@ -2739,7 +2752,7 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
 
             if ((0 == numBatchesExecuted) && !isInternalEncryptionQuery && connection.isAEv2()
                     && !encryptionMetadataIsRetrieved) {
-                this.enclaveCEKs = connection.initEnclaveParameters(preparedSQL, preparedTypeDefinitions, batchParam,
+                this.enclaveCEKs = connection.initEnclaveParameters(this, preparedSQL, preparedTypeDefinitions, batchParam,
                         parameterNames);
                 encryptionMetadataIsRetrieved = true;
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java
index 278197bc..02224d0a 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java
@@ -153,7 +153,7 @@ public final class SQLServerResource extends ListResourceBundle {
         {"R_sslRequiredNoServerSupport", "The driver could not establish a secure connection to SQL Server by using Secure Sockets Layer (SSL) encryption. The application requested encryption but the server is not configured to support SSL."},
         {"R_sslRequiredByServer", "SQL Server login requires an encrypted connection that uses Secure Sockets Layer (SSL)."},
         {"R_sslFailed", "The driver could not establish a secure connection to SQL Server by using Secure Sockets Layer (SSL) encryption. Error: \"{0}\"."},
-        {"R_certNameFailed", "Failed to validate the server name in a certificate during Secure Sockets Layer (SSL) initialization."},
+        {"R_certNameFailed", "Failed to validate the server name \"{0}\"in a certificate during Secure Sockets Layer (SSL) initialization. Name in certificate \"{1}\""},
         {"R_failedToInitializeXA", "Failed to initialize the stored procedure xp_sqljdbc_xa_init. The status is: {0}. Error: \"{1}\""},
         {"R_failedFunctionXA", "The function {0} failed. The status is: {1}. Error: \"{2}\""},
         {"R_noTransactionCookie", "The function {0} failed. No transaction cookie was returned."},
@@ -273,6 +273,8 @@ public final class SQLServerResource extends ListResourceBundle {
         {"R_KeyExtractionFailed", "Key extraction failed : {0} ."},
         {"R_UntrustedKeyPath", "The column master key path {0} received from server {1} is not a trusted key path. The column master key path may be corrupt or you should set {0} as a trusted key path using SQLServerConnection.setColumnEncryptionTrustedMasterKeyPaths()."},
         {"R_UnrecognizedKeyStoreProviderName", "Failed to decrypt a column encryption key. Invalid key store provider name: {0}. A key store provider name must denote either a system key store provider or a registered custom key store provider. Valid system key provider names are: {1}. Valid (currently registered) custom key store provider names are: {2}. Please verify key store provider information in column master key definitions in the database, and verify all custom key store providers used in your application are registered properly."},
+        {"R_UnrecognizedConnectionKeyStoreProviderName", "Failed to decrypt a column encryption key. Invalid key store provider name: {0}. A key store provider name must denote a registered key store provider. Registered key store provider names on this connection are: {1}. Please verify key store provider information in column master key definitions in the database, and verify all key store providers used in your application are registered properly."},
+        {"R_UnrecognizedStatementKeyStoreProviderName", "Failed to decrypt a column encryption key. Invalid key store provider name: {0}. A key store provider name must denote a registered key store provider. Registered key store provider names on this statement are: {1}. Please verify key store provider information in column master key definitions in the database, and verify all key store providers used in your application are registered properly."},
         {"R_UnsupportedDataTypeAE", "Encryption and decryption of data type {0} is not supported."},
         {"R_NormalizationErrorAE", "Decryption of the data type {0} failed. Normalization error."},
         {"R_UnsupportedNormalizationVersionAE", "Normalization version \"{0}\" received from SQL Server is either invalid or corrupted. Valid normalization versions are: {1}."},
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResultSet.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResultSet.java
index 8ded16e5..9f2ca277 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResultSet.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResultSet.java
@@ -2074,7 +2074,7 @@ public class SQLServerResultSet implements ISQLServerResultSet, java.io.Serializ
 
     private Object getValue(int columnIndex, JDBCType jdbcType, InputStreamGetterArgs getterArgs,
             Calendar cal) throws SQLServerException {
-        Object o = getterGetColumn(columnIndex).getValue(jdbcType, getterArgs, cal, tdsReader);
+        Object o = getterGetColumn(columnIndex).getValue(jdbcType, getterArgs, cal, tdsReader, stmt);
         lastValueWasNull = (null == o);
         return o;
     }
@@ -3083,7 +3083,7 @@ public class SQLServerResultSet implements ISQLServerResultSet, java.io.Serializ
         assert null != tdsReader;
 
         return deletedCurrentRow
-                || (0 != serverCursorId && TDS.ROWSTAT_FETCH_MISSING == loadColumn(columns.length).getInt(tdsReader));
+                || (0 != serverCursorId && TDS.ROWSTAT_FETCH_MISSING == loadColumn(columns.length).getInt(tdsReader, stmt));
     }
 
     /* ---------------- Column updates ---------------------- */
@@ -4777,7 +4777,7 @@ public class SQLServerResultSet implements ISQLServerResultSet, java.io.Serializ
             tdsWriter.writeRPCStringUnicode(tableName);
 
             for (Column column : columns)
-                column.sendByRPC(tdsWriter, stmt.connection);
+                column.sendByRPC(tdsWriter, stmt);
         } else {
             tdsWriter.writeRPCStringUnicode("");
             tdsWriter.writeRPCStringUnicode("INSERT INTO " + tableName + " DEFAULT VALUES");
@@ -4858,7 +4858,7 @@ public class SQLServerResultSet implements ISQLServerResultSet, java.io.Serializ
         assert hasUpdatedColumns();
 
         for (Column column : columns)
-            column.sendByRPC(tdsWriter, stmt.connection);
+            column.sendByRPC(tdsWriter, stmt);
 
         TDSParser.parse(command.startResponse(), command.getLogContext());
     }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSecurityUtility.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSecurityUtility.java
index f75e6550..d0b64a4f 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSecurityUtility.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSecurityUtility.java
@@ -44,6 +44,8 @@ class SQLServerSecurityUtility {
     static final int INTERNAL_SERVER_ERROR = 500;
     static final int NETWORK_CONNECT_TIMEOUT_ERROR = 599;
 
+    static final String WINDOWS_KEY_STORE_NAME = "MSSQL_CERTIFICATE_STORE";
+
     /**
      * Give the hash of given plain text
      * 
@@ -96,17 +98,73 @@ class SQLServerSecurityUtility {
 
     }
 
+    static SQLServerColumnEncryptionKeyStoreProvider getColumnEncryptionKeyStoreProvider(String providerName, SQLServerConnection connection, SQLServerStatement statement) throws SQLServerException {
+        assert providerName != null && providerName.length() != 0 : "Provider name should not be null or empty";
+
+        // check statement level KeyStoreProvider if statement is not null.
+        if (statement != null && statement.hasColumnEncryptionKeyStoreProvidersRegistered()) {
+            return statement.getColumnEncryptionKeyStoreProvider(providerName);
+        }
+
+        return connection.getColumnEncryptionKeyStoreProviderOnConnection(providerName);
+    }
+
+    static boolean shouldUseInstanceLevelProviderFlow(String keyStoreName, SQLServerConnection connection, SQLServerStatement statement) {
+        return !keyStoreName.equalsIgnoreCase(WINDOWS_KEY_STORE_NAME) 
+            && (connection.hasConnectionColumnEncryptionKeyStoreProvidersRegistered() || (null != statement && statement.hasColumnEncryptionKeyStoreProvidersRegistered()));   
+    }
+
+    static SQLServerSymmetricKey getKeyFromLocalProviders(EncryptionKeyInfo keyInfo, SQLServerConnection connection, SQLServerStatement statement) throws SQLServerException {
+        String serverName = connection.getTrustedServerNameAE();
+        assert null != serverName : "serverName should not be null in getKey.";
+
+        if (connectionlogger.isLoggable(java.util.logging.Level.FINE)) {
+            connectionlogger.fine("Checking trusted master key path...");
+        }
+        Boolean[] hasEntry = new Boolean[1];
+        List<String> trustedKeyPaths = SQLServerConnection.getColumnEncryptionTrustedMasterKeyPaths(serverName, hasEntry);
+        if (hasEntry[0]) {
+            if ((null == trustedKeyPaths) || (0 == trustedKeyPaths.size()) || (!trustedKeyPaths.contains(keyInfo.keyPath))) {
+                MessageFormat form = new MessageFormat(SQLServerException.getErrString("R_UntrustedKeyPath"));
+                Object[] msgArgs = {keyInfo.keyPath, serverName};
+                throw new SQLServerException(null, form.format(msgArgs), null, 0, false);
+            }
+        }
+
+        SQLServerException lastException = null;
+        SQLServerColumnEncryptionKeyStoreProvider provider = null;
+        byte[] plaintextKey = null;
+        
+        try {
+            provider = getColumnEncryptionKeyStoreProvider(keyInfo.keyStoreName, connection, statement);
+            plaintextKey = provider.decryptColumnEncryptionKey(keyInfo.keyPath, keyInfo.algorithmName, keyInfo.encryptedKey);
+            
+        } catch (SQLServerException e) {
+            lastException = e;
+        }
+        
+        if (null == plaintextKey) {
+            if (null != lastException) {
+                throw lastException;
+            } else {
+                throw new SQLServerException(null, SQLServerException.getErrString("R_CEKDecryptionFailed"), null, 0, false);
+            }
+        }
+        
+        return new SQLServerSymmetricKey(plaintextKey);
+    }
+
     /*
      * Encrypts the ciphertext.
      */
     static byte[] encryptWithKey(byte[] plainText, CryptoMetadata md,
-            SQLServerConnection connection) throws SQLServerException {
+            SQLServerConnection connection, SQLServerStatement statement) throws SQLServerException {
         String serverName = connection.getTrustedServerNameAE();
-        assert serverName != null : "Server name should npt be null in EncryptWithKey";
+        assert serverName != null : "Server name should not be null in EncryptWithKey";
 
         // Initialize cipherAlgo if not already done.
         if (!md.IsAlgorithmInitialized()) {
-            SQLServerSecurityUtility.decryptSymmetricKey(md, connection);
+            SQLServerSecurityUtility.decryptSymmetricKey(md, connection, statement);
         }
 
         assert md.IsAlgorithmInitialized();
@@ -144,21 +202,26 @@ class SQLServerSecurityUtility {
      *        The cipher metadata
      * @param connection
      *        The connection
+     * @param statement
+     *        The statemenet
      */
-    static void decryptSymmetricKey(CryptoMetadata md, SQLServerConnection connection) throws SQLServerException {
+    static void decryptSymmetricKey(CryptoMetadata md, SQLServerConnection connection, SQLServerStatement statement) throws SQLServerException {
         assert null != md : "md should not be null in DecryptSymmetricKey.";
         assert null != md.cekTableEntry : "md.EncryptionInfo should not be null in DecryptSymmetricKey.";
         assert null != md.cekTableEntry.columnEncryptionKeyValues : "md.EncryptionInfo.ColumnEncryptionKeyValues should not be null in DecryptSymmetricKey.";
 
         SQLServerSymmetricKey symKey = null;
         EncryptionKeyInfo encryptionkeyInfoChosen = null;
-        SQLServerSymmetricKeyCache cache = SQLServerSymmetricKeyCache.getInstance();
+        SQLServerSymmetricKeyCache globalCEKCache = SQLServerSymmetricKeyCache.getInstance();
         Iterator<EncryptionKeyInfo> it = md.cekTableEntry.columnEncryptionKeyValues.iterator();
         SQLServerException lastException = null;
         while (it.hasNext()) {
             EncryptionKeyInfo keyInfo = it.next();
             try {
-                symKey = cache.getKey(keyInfo, connection);
+                symKey = shouldUseInstanceLevelProviderFlow(keyInfo.keyStoreName, connection, statement) ?
+                getKeyFromLocalProviders(keyInfo, connection, statement) :
+                globalCEKCache.getKey(keyInfo, connection);
+
                 if (null != symKey) {
                     encryptionkeyInfoChosen = keyInfo;
                     break;
@@ -197,13 +260,13 @@ class SQLServerSecurityUtility {
      * Decrypts the ciphertext.
      */
     static byte[] decryptWithKey(byte[] cipherText, CryptoMetadata md,
-            SQLServerConnection connection) throws SQLServerException {
+            SQLServerConnection connection, SQLServerStatement statement) throws SQLServerException {
         String serverName = connection.getTrustedServerNameAE();
         assert null != serverName : "serverName should not be null in DecryptWithKey.";
 
         // Initialize cipherAlgo if not already done.
         if (!md.IsAlgorithmInitialized()) {
-            SQLServerSecurityUtility.decryptSymmetricKey(md, connection);
+            SQLServerSecurityUtility.decryptSymmetricKey(md, connection, statement);
         }
 
         assert md.IsAlgorithmInitialized() : "Decryption Algorithm is not initialized";
@@ -218,7 +281,7 @@ class SQLServerSecurityUtility {
     /*
      * Verify the signature for the CMK
      */
-    static void verifyColumnMasterKeyMetadata(SQLServerConnection connection, String keyStoreName, String keyPath,
+    static void verifyColumnMasterKeyMetadata(SQLServerConnection connection, SQLServerStatement statement, String keyStoreName, String keyPath,
             String serverName, boolean isEnclaveEnabled, byte[] CMKSignature) throws SQLServerException {
 
         // check trusted key paths
@@ -233,8 +296,14 @@ class SQLServerSecurityUtility {
             }
         }
 
-        if (!connection.getColumnEncryptionKeyStoreProvider(keyStoreName).verifyColumnMasterKeyMetadata(keyPath,
-                isEnclaveEnabled, CMKSignature)) {
+        SQLServerColumnEncryptionKeyStoreProvider provider = null;
+        if (shouldUseInstanceLevelProviderFlow(keyStoreName, connection, statement)) {
+            provider = getColumnEncryptionKeyStoreProvider(keyStoreName, connection, statement);
+        } else {
+            provider = connection.getSystemOrGlobalColumnEncryptionKeyStoreProvider(keyStoreName);
+        }
+
+        if (!provider.verifyColumnMasterKeyMetadata(keyPath, isEnclaveEnabled, CMKSignature)) {
             throw new SQLServerException(SQLServerException.getErrString("R_VerifySignature"), null);
         }
     }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java
index afc97fa8..f0480d4d 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java
@@ -16,6 +16,8 @@ import java.sql.SQLWarning;
 import java.sql.Statement;
 import java.text.MessageFormat;
 import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Stack;
 import java.util.StringTokenizer;
 import java.util.Vector;
@@ -72,6 +74,8 @@ public class SQLServerStatement implements ISQLServerStatement {
     }
 
     final static String identityQuery = " select SCOPE_IDENTITY() AS GENERATED_KEYS";
+    
+    static final String WINDOWS_KEY_STORE_NAME = "MSSQL_CERTIFICATE_STORE";
 
     /** the stored procedure name to call (if there is one) */
     String procedureName;
@@ -115,7 +119,7 @@ public class SQLServerStatement implements ISQLServerStatement {
     final SQLServerConnection connection;
 
     /**
-     * The user's specifed query timeout (in seconds).
+     * The user's specified query timeout (in seconds).
      */
     int queryTimeout;
 
@@ -135,11 +139,11 @@ public class SQLServerStatement implements ISQLServerStatement {
      * cancellation through Statement.cancel.
      *
      * Note: currentCommand is declared volatile to ensure that the JVM always returns the most recently set value for
-     * currentCommand to the cancelling thread.
+     * currentCommand to the canceling thread.
      */
     private volatile TDSCommand currentCommand = null;
 
-    /** last statment exec command */
+    /** last statement exec command */
     private TDSCommand lastStmtExecCmd = null;
 
     final void discardLastExecutionResults() {
@@ -404,8 +408,10 @@ public class SQLServerStatement implements ISQLServerStatement {
     boolean expectCursorOutParams;
 
     class StmtExecOutParamHandler extends TDSTokenHandler {
-        StmtExecOutParamHandler() {
+        SQLServerStatement statement;
+        StmtExecOutParamHandler(SQLServerStatement statement) {
             super("StmtExecOutParamHandler");
+            this.statement = statement;
         }
 
         boolean onRetStatus(TDSReader tdsReader) throws SQLServerException {
@@ -420,13 +426,13 @@ public class SQLServerStatement implements ISQLServerStatement {
 
                 // Read the cursor ID
                 param.skipRetValStatus(tdsReader);
-                serverCursorId = param.getInt(tdsReader);
+                serverCursorId = param.getInt(tdsReader, statement);
                 param.skipValue(tdsReader, true);
 
                 param = new Parameter(Util.shouldHonorAEForParameters(stmtColumnEncriptionSetting, connection));
                 // Read the row count (-1 means unknown)
                 param.skipRetValStatus(tdsReader);
-                if (-1 == (serverCursorRowCount = param.getInt(tdsReader)))
+                if (-1 == (serverCursorRowCount = param.getInt(tdsReader, statement)))
                     serverCursorRowCount = SQLServerResultSet.UNKNOWN_ROW_COUNT;
                 param.skipValue(tdsReader, true);
 
@@ -847,7 +853,7 @@ public class SQLServerStatement implements ISQLServerStatement {
         // call syntax is rewritten here as SQL exec syntax.
         String sql = ensureSQLSyntax(execCmd.sql);
         if (!isInternalEncryptionQuery && connection.isAEv2()) {
-            execCmd.enclaveCEKs = connection.initEnclaveParameters(sql, null, null, null);
+            execCmd.enclaveCEKs = connection.initEnclaveParameters(this, sql, null, null, null);
         }
 
         // If this request might be a query (as opposed to an update) then make
@@ -949,7 +955,7 @@ public class SQLServerStatement implements ISQLServerStatement {
 
         String batchStatementString = String.join(";", batchStatementBuffer);
         if (connection.isAEv2()) {
-            execCmd.enclaveCEKs = connection.initEnclaveParameters(batchStatementString, null, null, null);
+            execCmd.enclaveCEKs = connection.initEnclaveParameters(this, batchStatementString, null, null, null);
         }
 
         if (loggerExternal.isLoggable(Level.FINER) && Util.isActivityTraceOn()) {
@@ -1314,6 +1320,7 @@ public class SQLServerStatement implements ISQLServerStatement {
     final void processExecuteResults() throws SQLServerException {
         if (wasExecuted()) {
             processBatch();
+            checkClosed(); // processBatch could have resulted in a closed connection if isCloseOnCompletion is set
             TDSParser.parse(resultsReader(), "batch completion");
             ensureExecuteResultsReader(null);
         }
@@ -1712,7 +1719,7 @@ public class SQLServerStatement implements ISQLServerStatement {
      */
     boolean consumeExecOutParam(TDSReader tdsReader) throws SQLServerException {
         if (expectCursorOutParams) {
-            TDSParser.parse(tdsReader, new StmtExecOutParamHandler());
+            TDSParser.parse(tdsReader, new StmtExecOutParamHandler(this));
             return true;
         }
 
@@ -2387,6 +2394,91 @@ public class SQLServerStatement implements ISQLServerStatement {
         loggerExternal.exiting(getClassNameLogging(), "getResponseBuffering", responseBuff);
         return responseBuff;
     }
+
+    /** This is a per-statement store provider. */
+    Map<String, SQLServerColumnEncryptionKeyStoreProvider> statementColumnEncryptionKeyStoreProviders = new HashMap<>();
+
+    /**
+     * Registers statement-level key store providers, replacing all existing providers.
+     * 
+     * @param clientKeyStoreProviders
+     *        a map containing the store providers information.
+     * @throws SQLServerException
+     *         when an error occurs
+     */
+    public synchronized void registerColumnEncryptionKeyStoreProvidersOnStatement(
+            Map<String, SQLServerColumnEncryptionKeyStoreProvider> clientKeyStoreProviders) throws SQLServerException {
+        loggerExternal.entering(loggingClassName, "registerColumnEncryptionKeyStoreProvidersOnStatement",
+                "Registering Column Encryption Key Store Providers on Statement");
+
+        checkClosed();
+
+        if (null == clientKeyStoreProviders) {
+            throw new SQLServerException(null, SQLServerException.getErrString("R_CustomKeyStoreProviderMapNull"), null,
+                    0, false);
+        }
+
+        statementColumnEncryptionKeyStoreProviders.clear();
+
+        for (Map.Entry<String, SQLServerColumnEncryptionKeyStoreProvider> entry : clientKeyStoreProviders.entrySet()) {
+            String providerName = entry.getKey();
+            if (null == providerName || 0 == providerName.trim().length()) {
+                throw new SQLServerException(null, SQLServerException.getErrString("R_EmptyCustomKeyStoreProviderName"),
+                        null, 0, false);
+            }
+
+            // MSSQL_CERTIFICATE_STORE not allowed on statement level
+            if ((providerName.equalsIgnoreCase(WINDOWS_KEY_STORE_NAME))) {
+                MessageFormat form = new MessageFormat(
+                        SQLServerException.getErrString("R_InvalidCustomKeyStoreProviderName"));
+                Object[] msgArgs = {providerName, WINDOWS_KEY_STORE_NAME};
+                throw new SQLServerException(null, form.format(msgArgs), null, 0, false);
+            }
+
+            if (null == entry.getValue()) {
+                throw new SQLServerException(null,
+                        String.format(SQLServerException.getErrString("R_CustomKeyStoreProviderValueNull"), providerName),
+                        null, 0, false);
+            }
+
+            statementColumnEncryptionKeyStoreProviders.put(entry.getKey(), entry.getValue());
+        }
+
+        loggerExternal.exiting(loggingClassName, "registerColumnEncryptionKeyStoreProvidersOnStatement",
+                "Number of statement-level Key store providers that are registered: "
+                        + statementColumnEncryptionKeyStoreProviders.size());
+    }
+
+    synchronized String getAllStatementColumnEncryptionKeyStoreProviders() {
+        String keyStores = "";
+        if (0 != statementColumnEncryptionKeyStoreProviders.size())
+            keyStores = statementColumnEncryptionKeyStoreProviders.keySet().toString();
+        return keyStores;
+    }
+
+    synchronized boolean hasColumnEncryptionKeyStoreProvidersRegistered() {
+        return null != statementColumnEncryptionKeyStoreProviders && statementColumnEncryptionKeyStoreProviders.size() > 0;
+    }
+
+    synchronized SQLServerColumnEncryptionKeyStoreProvider getColumnEncryptionKeyStoreProvider(
+            String providerName) throws SQLServerException {
+
+        // Check for a statement-level provider first
+        if (null != statementColumnEncryptionKeyStoreProviders &&
+                statementColumnEncryptionKeyStoreProviders.size() > 0) {
+            // If any statement-level providers are registered, we don't fall back to connection-level providers
+            if (statementColumnEncryptionKeyStoreProviders.containsKey(providerName)) {
+                return statementColumnEncryptionKeyStoreProviders.get(providerName);
+            } else {
+                MessageFormat form = new MessageFormat(
+                        SQLServerException.getErrString("R_UnrecognizedStatementKeyStoreProviderName"));
+                Object[] msgArgs = {providerName, getAllStatementColumnEncryptionKeyStoreProviders()};
+                throw new SQLServerException(form.format(msgArgs), null);
+            }
+        }
+
+        return null;
+    }
 }
 
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSymmetricKeyCache.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSymmetricKeyCache.java
index c16c0aec..c88490b8 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSymmetricKeyCache.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerSymmetricKeyCache.java
@@ -9,40 +9,9 @@ import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 import java.text.MessageFormat;
+import java.time.Duration;
 import java.util.Base64;
 import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
-
-
-class CacheClear implements Runnable {
-
-    private String keylookupValue;
-    static private java.util.logging.Logger aeLogger = java.util.logging.Logger
-            .getLogger("com.microsoft.sqlserver.jdbc.CacheClear");
-
-    CacheClear(String keylookupValue) {
-        this.keylookupValue = keylookupValue;
-    }
-
-    @Override
-    public void run() {
-        // remove() is a no-op if the key is not in the map.
-        // It is a concurrentHashMap, update/remove operations are thread safe.
-        synchronized (SQLServerSymmetricKeyCache.lock) {
-            SQLServerSymmetricKeyCache instance = SQLServerSymmetricKeyCache.getInstance();
-            if (instance.getCache().containsKey(keylookupValue)) {
-                instance.getCache().get(keylookupValue).zeroOutKey();
-                instance.getCache().remove(keylookupValue);
-                if (aeLogger.isLoggable(java.util.logging.Level.FINE)) {
-                    aeLogger.fine("Removed encryption key from cache...");
-                }
-            }
-        }
-    }
-}
 
 
 /**
@@ -52,29 +21,21 @@ class CacheClear implements Runnable {
  */
 final class SQLServerSymmetricKeyCache {
     static final Object lock = new Object();
-    private final ConcurrentHashMap<String, SQLServerSymmetricKey> cache;
+    private final SimpleTtlCache<String, SQLServerSymmetricKey> cache;
     private static final SQLServerSymmetricKeyCache instance = new SQLServerSymmetricKeyCache();
-    private static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, new ThreadFactory() {
-        @Override
-        public Thread newThread(Runnable r) {
-            Thread t = Executors.defaultThreadFactory().newThread(r);
-            t.setDaemon(true);
-            return t;
-        }
-    });
 
     static final private java.util.logging.Logger aeLogger = java.util.logging.Logger
             .getLogger("com.microsoft.sqlserver.jdbc.SQLServerSymmetricKeyCache");
 
     private SQLServerSymmetricKeyCache() {
-        cache = new ConcurrentHashMap<>();
+        cache = new SimpleTtlCache<String, SQLServerSymmetricKey>();
     }
 
     static SQLServerSymmetricKeyCache getInstance() {
         return instance;
     }
 
-    ConcurrentHashMap<String, SQLServerSymmetricKey> getCache() {
+    SimpleTtlCache<String, SQLServerSymmetricKey> getCache() {
         return cache;
     }
 
@@ -123,11 +84,21 @@ final class SQLServerSymmetricKeyCache {
                 aeLogger.fine("Checking Symmetric key cache...");
             }
 
-            // if ColumnEncryptionKeyCacheTtl is 0 no caching at all
-            if (!cache.containsKey(keyLookupValue)) {
+            if (!cache.contains(keyLookupValue)) {
+ 
+                // search system/global key store providers
+                SQLServerColumnEncryptionKeyStoreProvider provider = connection.getSystemOrGlobalColumnEncryptionKeyStoreProvider(keyInfo.keyStoreName);
+                assert null != provider : "Provider should not be null.";
+
                 byte[] plaintextKey;
-                plaintextKey = connection.getColumnEncryptionKeyStoreProvider(keyInfo.keyStoreName)
-                        .decryptColumnEncryptionKey(keyInfo.keyPath, keyInfo.algorithmName, keyInfo.encryptedKey);
+                
+                /* 
+                 * When provider decrypt Column Encryption Key, it can cache the decrypted key if cacheTTL > 0.
+                 * To prevent conflicts between CEK caches, system providers and global providers should not use their own CEK caches.
+                 */
+                provider.setColumnEncryptionCacheTtl(Duration.ZERO);
+                plaintextKey = provider.decryptColumnEncryptionKey(keyInfo.keyPath, keyInfo.algorithmName, keyInfo.encryptedKey);
+
                 encryptionKey = new SQLServerSymmetricKey(plaintextKey);
 
                 /*
@@ -137,11 +108,8 @@ final class SQLServerSymmetricKeyCache {
                  */
                 long columnEncryptionKeyCacheTtl = SQLServerConnection.getColumnEncryptionKeyCacheTtl();
                 if (0 != columnEncryptionKeyCacheTtl) {
-                    cache.putIfAbsent(keyLookupValue, encryptionKey);
-                    if (aeLogger.isLoggable(java.util.logging.Level.FINE)) {
-                        aeLogger.fine("Adding encryption key to cache...");
-                    }
-                    scheduler.schedule(new CacheClear(keyLookupValue), columnEncryptionKeyCacheTtl, SECONDS);
+                    cache.setCacheTtl(columnEncryptionKeyCacheTtl);
+                    cache.put(keyLookupValue, encryptionKey);
                 }
             } else {
                 encryptionKey = cache.get(keyLookupValue);
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerVSMEnclaveProvider.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerVSMEnclaveProvider.java
index f04284a5..d7a177f9 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerVSMEnclaveProvider.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerVSMEnclaveProvider.java
@@ -55,7 +55,7 @@ public class SQLServerVSMEnclaveProvider implements ISQLServerEnclaveProvider {
     }
 
     @Override
-    public ArrayList<byte[]> createEnclaveSession(SQLServerConnection connection, String userSql,
+    public ArrayList<byte[]> createEnclaveSession(SQLServerConnection connection, SQLServerStatement statement, String userSql,
             String preparedTypeDefinitions, Parameter[] params,
             ArrayList<String> parameterNames) throws SQLServerException {
         // Check if the session exists in our cache
@@ -66,7 +66,7 @@ public class SQLServerVSMEnclaveProvider implements ISQLServerEnclaveProvider {
             this.enclaveSession = entry.getEnclaveSession();
             this.vsmParams = (VSMAttestationParameters) entry.getBaseAttestationRequest();
         }
-        ArrayList<byte[]> b = describeParameterEncryption(connection, userSql, preparedTypeDefinitions, params,
+        ArrayList<byte[]> b = describeParameterEncryption(connection, statement, userSql, preparedTypeDefinitions, params,
                 parameterNames);
         if (connection.enclaveEstablished()) {
             return b;
@@ -141,7 +141,7 @@ public class SQLServerVSMEnclaveProvider implements ISQLServerEnclaveProvider {
         return certData;
     }
 
-    private ArrayList<byte[]> describeParameterEncryption(SQLServerConnection connection, String userSql,
+    private ArrayList<byte[]> describeParameterEncryption(SQLServerConnection connection, SQLServerStatement statement, String userSql,
             String preparedTypeDefinitions, Parameter[] params,
             ArrayList<String> parameterNames) throws SQLServerException {
         ArrayList<byte[]> enclaveRequestedCEKs = new ArrayList<>();
@@ -153,7 +153,7 @@ public class SQLServerVSMEnclaveProvider implements ISQLServerEnclaveProvider {
                     // Should never happen.
                     return enclaveRequestedCEKs;
                 }
-                processSDPEv1(userSql, preparedTypeDefinitions, params, parameterNames, connection, stmt, rs,
+                processSDPEv1(userSql, preparedTypeDefinitions, params, parameterNames, connection, statement, stmt, rs,
                         enclaveRequestedCEKs);
                 // Process the third resultset.
                 if (connection.isAEv2() && stmt.getMoreResults()) {
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SSPIAuthentication.java b/src/main/java/com/microsoft/sqlserver/jdbc/SSPIAuthentication.java
index c0bbe0d7..f8f5659b 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SSPIAuthentication.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SSPIAuthentication.java
@@ -33,7 +33,7 @@ abstract class SSPIAuthentication {
      */
     private static final Pattern SPN_PATTERN = Pattern.compile("MSSQLSvc/(.*):([^:@]+)(@.+)?",
             Pattern.CASE_INSENSITIVE);
-    
+
     private static final Logger logger = Logger.getLogger("com.microsoft.sqlserver.jdbc.SSPIAuthentication");
 
     /**
@@ -141,9 +141,14 @@ abstract class SSPIAuthentication {
             // Realm is already present, no need to enrich, the job has already been done
             return spn;
         }
+
+        // Try to derive realm if not specified in the connection. This might take some time if DNS lookup is slow
+        if (logger.isLoggable(Level.FINER)) {
+            logger.finer("Deriving realm");
+        }
+
         String dnsName = m.group(1);
         String portOrInstance = m.group(2);
-        // If realm is not specified in the connection, try to derive it.
         if (null == realm || realm.trim().isEmpty()) {
             RealmValidator realmValidator = getRealmValidator();
             realm = findRealmFromHostname(realmValidator, dnsName);
@@ -152,7 +157,7 @@ abstract class SSPIAuthentication {
                 try {
                     String canonicalHostName = InetAddress.getByName(dnsName).getCanonicalHostName();
                     realm = findRealmFromHostname(realmValidator, canonicalHostName);
-                    // match means hostname is correct (for instance if server name was an IP) so override dnsName as well
+                    // match means hostname is correct (eg if server name was an IP) so override dnsName as well
                     dnsName = canonicalHostName;
                 } catch (UnknownHostException e) {
                     // ignored, cannot canonicalize
@@ -162,9 +167,16 @@ abstract class SSPIAuthentication {
                 }
             }
         }
+
         if (null == realm) {
+            if (logger.isLoggable(Level.FINER)) {
+                logger.finer("Could not derive realm.");
+            }
             return spn;
         } else {
+            if (logger.isLoggable(Level.FINER)) {
+                logger.finer("Derived realm: " + realm);
+            }
             StringBuilder sb = new StringBuilder("MSSQLSvc/");
             sb.append(dnsName).append(":").append(portOrInstance).append("@").append(realm.toUpperCase(Locale.ENGLISH));
             return sb.toString();
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SimpleTtlCache.java b/src/main/java/com/microsoft/sqlserver/jdbc/SimpleTtlCache.java
new file mode 100644
index 00000000..a93f5a92
--- /dev/null
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SimpleTtlCache.java
@@ -0,0 +1,173 @@
+/*
+ * Microsoft JDBC Driver for SQL Server Copyright(c) Microsoft Corporation All rights reserved. This program is made
+ * available under the terms of the MIT License. See the LICENSE file in the project root for more information.
+ */
+
+package com.microsoft.sqlserver.jdbc;
+
+import java.time.Duration;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+/**
+ * Internal class implements a cache data structure based on ConcurrentHashMap. The cache uses a scheduler to 
+ * provide eviction for the cache entry according to the pre-defined time-to-live value.
+ * 
+ * This class is used as local caches for the column encrypted keys and the results of signature verification of 
+ * column master key metadata. Refer to SQLServerColumnEncryptionAzureKeyVaultProvider for an example.
+ */
+final class SimpleTtlCache<K,V> {    
+
+    // This class clears cache entry when it is called by scheduler. 
+    class CacheClear implements Runnable {
+    
+        private K keylookupValue;
+        final private java.util.logging.Logger logger = java.util.logging.Logger
+            .getLogger("com.microsoft.sqlserver.jdbc.SimpleTtlCache.CacheClear");
+    
+        CacheClear(K keylookupValue) {
+            this.keylookupValue = keylookupValue;
+        }
+    
+        @Override
+        public void run() {
+            // remove() is a no-op if the key is not in the map.
+            // It is a concurrentHashMap, update/remove operations are thread safe.
+            if (cache.containsKey(keylookupValue)) {
+                V value = cache.get(keylookupValue);
+                
+                if (value instanceof SQLServerSymmetricKey) {
+                    ((SQLServerSymmetricKey) value).zeroOutKey();
+                }
+                
+                cache.remove(keylookupValue);
+                if (logger.isLoggable(java.util.logging.Level.FINE)) {
+                    logger.fine("Removed key from cache...");
+                }
+            }
+        }
+    }
+
+    private static final java.util.logging.Logger simpleCacheLogger = java.util.logging.Logger.getLogger("com.microsoft.sqlserver.jdbc.SimpleTtlCache");
+    private static final long defaultTTLInHours = 2;  
+
+    private final ConcurrentHashMap<K, V> cache;
+    private Duration cacheTtl = Duration.ofHours(defaultTTLInHours);  // The default time-to-live is set to 2 hours.
+
+    private static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, new ThreadFactory() {
+        @Override
+        public Thread newThread(Runnable r) {
+            Thread t = Executors.defaultThreadFactory().newThread(r);
+            t.setDaemon(true);
+            return t;
+        }
+    });
+
+    /**
+     * Get the cache size.
+     * 
+     * @return cache size
+     */
+    int getCacheSize() {
+        return cache.size();
+    }
+
+    /**
+     * Set time-to-live value.
+     * 
+     * @param duration
+     *        time-to-live value in duration
+     */
+    void setCacheTtl(Duration duration) {
+        cacheTtl = duration;
+    }
+
+    /**
+     * Set time-to-live value.
+     * 
+     * @param seconds
+     *        time-to-live value in seconds
+     */
+    void setCacheTtl(long seconds) {
+        cacheTtl = Duration.ofSeconds(seconds);
+    }
+
+    /**
+     * Get the time-to-live value.
+     * 
+     * @return time-to-live in duration
+     */
+    Duration getCacheTtl() {
+        return cacheTtl;
+    }
+
+    /**
+     * Constructor used for creating a cache instance.
+     * 
+     * Use default time-to-live value.
+     */
+    SimpleTtlCache() {
+        cache = new ConcurrentHashMap<>();
+    }
+
+    /**
+     * Constructor used for creating a cache instance.
+     * 
+     * @param duration
+     *        time-to-live in duration
+     */
+    SimpleTtlCache(Duration duration) {
+        cacheTtl = duration;
+        cache = new ConcurrentHashMap<>();
+    }
+
+    /**
+     * Check if key exists.
+     * 
+     * @param key
+     *        key
+     * @return true if existed or false if not.
+     */
+    boolean contains(Object key) {
+        return cache.containsKey(key);
+    }
+
+    /**
+     * Get value from cache based on the key given.
+     * 
+     * @param key
+     *        key
+     * @return value
+     */
+    V get(Object key) {
+        return cache.get(key);
+    }
+
+    /**
+     * Put (Key, Value) entry into cache.
+     * 
+     * @param key
+     *        key
+     * @param value
+     *        value
+     * @return value
+     */
+    V put(K key, V value) {
+        V previousValue = null;
+        long cacheTtlInSeconds = cacheTtl.getSeconds();
+        
+        if (0 < cacheTtlInSeconds) {
+            previousValue = cache.put(key, value);
+            if (simpleCacheLogger.isLoggable(java.util.logging.Level.FINEST)) {
+                simpleCacheLogger.fine("Adding encryption key to cache...");
+            }
+            scheduler.schedule(new CacheClear(key), cacheTtlInSeconds, SECONDS);
+        }
+
+        return previousValue;
+    }
+
+}
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/dtv.java b/src/main/java/com/microsoft/sqlserver/jdbc/dtv.java
index c48cd47f..9c5c7f73 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/dtv.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/dtv.java
@@ -241,10 +241,10 @@ final class DTV {
      * determine the value (e.g. a Calendar object for time-valued DTV values).
      */
     Object getValue(JDBCType jdbcType, int scale, InputStreamGetterArgs streamGetterArgs, Calendar cal,
-            TypeInfo typeInfo, CryptoMetadata cryptoMetadata, TDSReader tdsReader) throws SQLServerException {
+            TypeInfo typeInfo, CryptoMetadata cryptoMetadata, TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException {
         if (null == impl)
             impl = new ServerDTVImpl();
-        return impl.getValue(this, jdbcType, scale, streamGetterArgs, cal, typeInfo, cryptoMetadata, tdsReader);
+        return impl.getValue(this, jdbcType, scale, streamGetterArgs, cal, typeInfo, cryptoMetadata, tdsReader, statement);
     }
 
     Object getSetterValue() {
@@ -271,9 +271,10 @@ final class DTV {
         private final boolean isOutParam;
         private final TDSWriter tdsWriter;
         private final SQLServerConnection conn;
+        private final SQLServerStatement statement;
 
         SendByRPCOp(String name, TypeInfo typeInfo, SQLCollation collation, int precision, int outScale,
-                boolean isOutParam, TDSWriter tdsWriter, SQLServerConnection conn) {
+                boolean isOutParam, TDSWriter tdsWriter, SQLServerStatement statement) {
             this.name = name;
             this.typeInfo = typeInfo;
             this.collation = collation;
@@ -281,7 +282,8 @@ final class DTV {
             this.outScale = outScale;
             this.isOutParam = isOutParam;
             this.tdsWriter = tdsWriter;
-            this.conn = conn;
+            this.conn = statement.connection;
+            this.statement = statement;
         }
 
         void execute(DTV dtv, String strValue) throws SQLServerException {
@@ -793,26 +795,26 @@ final class DTV {
                             case SMALLDATETIME:
                                 tdsWriter.writeEncryptedRPCDateTime(name,
                                         timestampNormalizedCalendar(calendar, javaType, conn.baseYear()),
-                                        subSecondNanos, isOutParam, jdbcType);
+                                        subSecondNanos, isOutParam, jdbcType, statement);
                                 break;
 
                             case TIMESTAMP:
                                 assert null != cryptoMeta;
                                 tdsWriter.writeEncryptedRPCDateTime2(name,
                                         timestampNormalizedCalendar(calendar, javaType, conn.baseYear()),
-                                        subSecondNanos, valueLength, isOutParam);
+                                        subSecondNanos, valueLength, isOutParam, statement);
                                 break;
 
                             case TIME:
                                 // when colum is encrypted, always send time as time, ignore sendTimeAsDatetime setting
                                 assert null != cryptoMeta;
                                 tdsWriter.writeEncryptedRPCTime(name, calendar, subSecondNanos, valueLength,
-                                        isOutParam);
+                                        isOutParam, statement);
                                 break;
 
                             case DATE:
                                 assert null != cryptoMeta;
-                                tdsWriter.writeEncryptedRPCDate(name, calendar, isOutParam);
+                                tdsWriter.writeEncryptedRPCDate(name, calendar, isOutParam, statement);
                                 break;
 
                             case TIMESTAMP_WITH_TIMEZONE:
@@ -830,7 +832,7 @@ final class DTV {
 
                                 assert null != cryptoMeta;
                                 tdsWriter.writeEncryptedRPCDateTimeOffset(name, calendar, minutesOffset, subSecondNanos,
-                                        valueLength, isOutParam);
+                                        valueLength, isOutParam, statement);
                                 break;
 
                             default:
@@ -865,15 +867,15 @@ final class DTV {
                                 if ((JDBCType.DATETIME == jdbcType) || (JDBCType.SMALLDATETIME == jdbcType)) {
                                     tdsWriter.writeEncryptedRPCDateTime(name,
                                             timestampNormalizedCalendar(calendar, javaType, conn.baseYear()),
-                                            subSecondNanos, isOutParam, jdbcType);
+                                            subSecondNanos, isOutParam, jdbcType, statement);
                                 } else if (0 == valueLength) {
                                     tdsWriter.writeEncryptedRPCDateTime2(name,
                                             timestampNormalizedCalendar(calendar, javaType, conn.baseYear()),
-                                            subSecondNanos, outScale, isOutParam);
+                                            subSecondNanos, outScale, isOutParam, statement);
                                 } else {
                                     tdsWriter.writeEncryptedRPCDateTime2(name,
                                             timestampNormalizedCalendar(calendar, javaType, conn.baseYear()),
-                                            subSecondNanos, (valueLength), isOutParam);
+                                            subSecondNanos, (valueLength), isOutParam, statement);
                                 }
                             } else
                                 tdsWriter.writeRPCDateTime2(name,
@@ -887,10 +889,10 @@ final class DTV {
                             if (null != cryptoMeta) {
                                 if (0 == valueLength) {
                                     tdsWriter.writeEncryptedRPCTime(name, calendar, subSecondNanos, outScale,
-                                            isOutParam);
+                                            isOutParam, statement);
                                 } else {
                                     tdsWriter.writeEncryptedRPCTime(name, calendar, subSecondNanos, valueLength,
-                                            isOutParam);
+                                            isOutParam, statement);
                                 }
                             } else {
                                 // Send the java.sql.Types.TIME value as TIME or DATETIME SQL Server
@@ -909,7 +911,7 @@ final class DTV {
 
                         case DATE:
                             if (null != cryptoMeta)
-                                tdsWriter.writeEncryptedRPCDate(name, calendar, isOutParam);
+                                tdsWriter.writeEncryptedRPCDate(name, calendar, isOutParam, statement);
                             else
                                 tdsWriter.writeRPCDate(name, calendar, isOutParam);
 
@@ -947,12 +949,12 @@ final class DTV {
                             if (null != cryptoMeta) {
                                 if (0 == valueLength) {
                                     tdsWriter.writeEncryptedRPCDateTimeOffset(name, calendar, minutesOffset,
-                                            subSecondNanos, outScale, isOutParam);
+                                            subSecondNanos, outScale, isOutParam, statement);
                                 } else {
                                     tdsWriter.writeEncryptedRPCDateTimeOffset(name, calendar, minutesOffset,
                                             subSecondNanos,
                                             (0 == valueLength ? TDS.MAX_FRACTIONAL_SECONDS_SCALE : valueLength),
-                                            isOutParam);
+                                            isOutParam, statement);
                                 }
                             } else
                                 tdsWriter.writeRPCDateTimeOffset(name, calendar, minutesOffset, subSecondNanos,
@@ -1115,7 +1117,7 @@ final class DTV {
             if (null != cryptoMeta) {
                 tdsWriter.writeRPCNameValType(name, isOutParam, TDSType.BIGVARBINARY);
                 if (null != byteArrayValue) {
-                    byteArrayValue = SQLServerSecurityUtility.encryptWithKey(byteArrayValue, cryptoMeta, conn);
+                    byteArrayValue = SQLServerSecurityUtility.encryptWithKey(byteArrayValue, cryptoMeta, conn, statement);
                     tdsWriter.writeEncryptedRPCByteArray(byteArrayValue);
                     writeEncryptData(dtv, false);
                 } else {
@@ -1910,9 +1912,9 @@ final class DTV {
      * Serializes a value as the specified type for RPC.
      */
     void sendByRPC(String name, TypeInfo typeInfo, SQLCollation collation, int precision, int outScale,
-            boolean isOutParam, TDSWriter tdsWriter, SQLServerConnection conn) throws SQLServerException {
+            boolean isOutParam, TDSWriter tdsWriter, SQLServerStatement statement) throws SQLServerException {
         // typeInfo is null when called from PreparedStatement->Parameter->SendByRPC
-        executeOp(new SendByRPCOp(name, typeInfo, collation, precision, outScale, isOutParam, tdsWriter, conn));
+        executeOp(new SendByRPCOp(name, typeInfo, collation, precision, outScale, isOutParam, tdsWriter, statement));
     }
 }
 
@@ -1953,7 +1955,7 @@ abstract class DTVImpl {
     abstract JavaType getJavaType();
 
     abstract Object getValue(DTV dtv, JDBCType jdbcType, int scale, InputStreamGetterArgs streamGetterArgs,
-            Calendar cal, TypeInfo type, CryptoMetadata cryptoMetadata, TDSReader tdsReader) throws SQLServerException;
+            Calendar cal, TypeInfo type, CryptoMetadata cryptoMetadata, TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException;
 
     abstract Object getSetterValue();
 
@@ -2328,7 +2330,7 @@ final class AppDTVImpl extends DTVImpl {
     }
 
     Object getValue(DTV dtv, JDBCType jdbcType, int scale, InputStreamGetterArgs streamGetterArgs, Calendar cal,
-            TypeInfo typeInfo, CryptoMetadata cryptoMetadata, TDSReader tdsReader) throws SQLServerException {
+            TypeInfo typeInfo, CryptoMetadata cryptoMetadata, TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException {
         // Client side type conversion is not supported
         if (this.jdbcType != jdbcType)
             DataTypes.throwConversionError(this.jdbcType.toString(), jdbcType.toString());
@@ -3658,7 +3660,7 @@ final class ServerDTVImpl extends DTVImpl {
     }
 
     Object getValue(DTV dtv, JDBCType jdbcType, int scale, InputStreamGetterArgs streamGetterArgs, Calendar cal,
-            TypeInfo typeInfo, CryptoMetadata cryptoMetadata, TDSReader tdsReader) throws SQLServerException {
+            TypeInfo typeInfo, CryptoMetadata cryptoMetadata, TDSReader tdsReader, SQLServerStatement statement) throws SQLServerException {
         SQLServerConnection con = tdsReader.getConnection();
         Object convertedValue = null;
         byte[] decryptedValue;
@@ -3734,7 +3736,7 @@ final class ServerDTVImpl extends DTVImpl {
                     throw new SQLServerException(SQLServerException.getErrString("R_notSupported"), null);
                 }
 
-                decryptedValue = SQLServerSecurityUtility.decryptWithKey((byte[]) convertedValue, cryptoMetadata, con);
+                decryptedValue = SQLServerSecurityUtility.decryptWithKey((byte[]) convertedValue, cryptoMetadata, con, statement);
                 return denormalizedValue(decryptedValue, jdbcType, cryptoMetadata.baseTypeInfo, con, streamGetterArgs,
                         cryptoMetadata.normalizationRuleVersion, cal);
             }
diff --git a/src/main/java/mssql/security/provider/MD4.java b/src/main/java/mssql/security/provider/MD4.java
index 463ee31f..7bc5d2cc 100644
--- a/src/main/java/mssql/security/provider/MD4.java
+++ b/src/main/java/mssql/security/provider/MD4.java
@@ -1,283 +1,312 @@
-package mssql.security.provider;
-
-/**
- * This code originates from sun.security.provider.MD4.java. It is modified to remove dependencies to DigestBase,
- * Provider, ByteArrayAccess.
- */
-
 /*
- * Copyright (c) 2005, 2016, Oracle and/or its affiliates. All rights reserved. ORACLE PROPRIETARY/CONFIDENTIAL. Use is
- * subject to license terms.
+ * Microsoft JDBC Driver for SQL Server Copyright(c) Microsoft Corporation All rights reserved. This program is made
+ * available under the terms of the MIT License. See the LICENSE file in the project root for more information.
  */
 
 /**
- * The MD4 class is used to compute an MD4 message digest over a given buffer of bytes. It is an implementation of the
- * RSA Data Security Inc MD4 algorithm as described in Internet RFC 1320.
- *
- * <p>
- * The MD4 algorithm is very weak and should not be used unless it is unavoidable. Therefore, it is not registered in
- * our standard providers. To obtain an implementation, call the static getInstance() method in this class.
- *
- * @author Andreas Sterbenz
+ * This code is extracted from org.bouncycastle.crypto.digests.MD4Digest and modified to remove dependencies.
  */
-public final class MD4 {
+package mssql.security.provider;
 
-    // state of this object
-    private final int[] state;
+public class MD4 {
+    private final byte[] xBuf = new byte[4];
+    private int xBufOff;
+    private long byteCount;
+    private static final int DIGEST_LENGTH = 16;
+    private int H1, H2, H3, H4; // IV's
+    private int[] X = new int[16];
+    private int xOff;
 
-    // temporary buffer, used by implCompress()
-    private final int[] x;
+    /**
+     * Standard constructor
+     */
+    public MD4() {
+        reset();
+    }
 
-    // rotation constants
-    private static final int S11 = 3;
-    private static final int S12 = 7;
-    private static final int S13 = 11;
-    private static final int S14 = 19;
-    private static final int S21 = 3;
-    private static final int S22 = 5;
-    private static final int S23 = 9;
-    private static final int S24 = 13;
-    private static final int S31 = 3;
-    private static final int S32 = 9;
-    private static final int S33 = 11;
-    private static final int S34 = 15;
+    /**
+     * Copy constructor. This will copy the state of the provided message digest.
+     */
+    public MD4(MD4 t) {
+        System.arraycopy(t.xBuf, 0, xBuf, 0, t.xBuf.length);
 
-    // size of the input to the compression function in bytes
-    private static final int blockSize = 64;
+        xBufOff = t.xBufOff;
+        byteCount = t.byteCount;
 
-    // buffer to store partial blocks, blockSize bytes large
-    private final byte[] buffer = new byte[blockSize];
-    // offset into buffer
-    private int bufOfs;
+        copyIn(t);
+    }
 
-    // number of bytes processed so far.
-    // also used as a flag to indicate reset status
-    // -1: need to call engineReset() before next call to update()
-    // 0: is already reset
-    private long bytesProcessed;
+    private void copyIn(MD4 t) {
+        System.arraycopy(t.xBuf, 0, xBuf, 0, t.xBuf.length);
 
-    private static final byte[] padding;
+        xBufOff = t.xBufOff;
+        byteCount = t.byteCount;
 
-    static {
-        padding = new byte[136];
-        padding[0] = (byte) 0x80;
+        H1 = t.H1;
+        H2 = t.H2;
+        H3 = t.H3;
+        H4 = t.H4;
+
+        System.arraycopy(t.X, 0, X, 0, t.X.length);
+        xOff = t.xOff;
     }
 
-    // Standard constructor, creates a new MD4 instance.
-    public MD4() {
-        state = new int[4];
-        x = new int[16];
-        implReset();
+    public String getAlgorithmName() {
+        return "MD4";
     }
 
-    /**
-     * Resets the digest for further use
-     */
-    public void reset() {
-        implReset();
+    public int getDigestSize() {
+        return DIGEST_LENGTH;
     }
 
-    /**
-     * Updates the digest using the specified array of bytes
-     */
-    public void update(byte[] input) {
-        engineUpdate(input, 0, input.length);
+    protected void processWord(byte[] in, int inOff) {
+        X[xOff++] = (in[inOff] & 0xff) | ((in[inOff + 1] & 0xff) << 8) | ((in[inOff + 2] & 0xff) << 16)
+                | ((in[inOff + 3] & 0xff) << 24);
+
+        if (xOff == 16) {
+            processBlock();
+        }
     }
 
-    /**
-     * Completes the hash computation by performing final operations such as padding.
-     */
-    public byte[] digest() {
-        byte[] out = new byte[16];
-        implDigest(out, 0);
-        return out;
+    protected void processLength(long bitLength) {
+        if (xOff > 14) {
+            processBlock();
+        }
+
+        X[14] = (int) (bitLength & 0xffffffff);
+        X[15] = (int) (bitLength >>> 32);
     }
 
-    /**
-     * Reset the state of this object.
-     */
-    private void implReset() {
-        // Load magic initialization constants.
-        state[0] = 0x67452301;
-        state[1] = 0xefcdab89;
-        state[2] = 0x98badcfe;
-        state[3] = 0x10325476;
-        bufOfs = 0;
-        bytesProcessed = 0;
+    private void unpackWord(int word, byte[] out, int outOff) {
+        out[outOff] = (byte) word;
+        out[outOff + 1] = (byte) (word >>> 8);
+        out[outOff + 2] = (byte) (word >>> 16);
+        out[outOff + 3] = (byte) (word >>> 24);
     }
 
-    /**
-     * Perform the final computations, any buffered bytes are added to the digest, the count is added to the digest, and
-     * the resulting digest is stored.
-     */
-    private void implDigest(byte[] out, int ofs) {
-        long bitsProcessed = bytesProcessed << 3;
-
-        int index = (int) bytesProcessed & 0x3f;
-        int padLen = (index < 56) ? (56 - index) : (120 - index);
-        engineUpdate(padding, 0, padLen);
-
-        /**
-         * the following replaces:
-         * 
-         * <pre>
-         * i2bLittle4((int) bitsProcessed, buffer, 56);
-         * i2bLittle4((int) (bitsProcessed >>> 32), buffer, 60);
-         * i2bLittle(state, 0, out, ofs, 16);
-         * </pre>
-         */
-        buffer[56] = (byte) bitsProcessed;
-        buffer[57] = (byte) (bitsProcessed >> 8);
-        buffer[58] = (byte) (bitsProcessed >> 16);
-        buffer[59] = (byte) (bitsProcessed >> 24);
-
-        buffer[60] = (byte) (bitsProcessed >> 32);
-        buffer[61] = (byte) (bitsProcessed >> 40);
-        buffer[62] = (byte) (bitsProcessed >> 48);
-        buffer[63] = (byte) (bitsProcessed >> 56);
-        implCompress(buffer, 0);
-
-        for (int i = 0; i < state.length; i++) {
-            int x = state[i];
-            out[ofs++] = (byte) x;
-            out[ofs++] = (byte) (x >> 8);
-            out[ofs++] = (byte) (x >> 16);
-            out[ofs++] = (byte) (x >> 24);
+    public void update(byte in) {
+        xBuf[xBufOff++] = in;
+
+        if (xBufOff == xBuf.length) {
+            processWord(xBuf, 0);
+            xBufOff = 0;
         }
+
+        byteCount++;
     }
 
-    private void engineUpdate(byte[] b, int ofs, int len) {
-        if (len == 0) {
-            return;
-        }
-        if ((ofs < 0) || (len < 0) || (ofs > b.length - len)) {
-            throw new ArrayIndexOutOfBoundsException();
-        }
-        if (bytesProcessed < 0) {
-            implReset();
-        }
-        bytesProcessed += len;
-        // if buffer is not empty, we need to fill it before proceeding
-        if (bufOfs != 0) {
-            int n = Math.min(len, blockSize - bufOfs);
-            System.arraycopy(b, ofs, buffer, bufOfs, n);
-            bufOfs += n;
-            ofs += n;
-            len -= n;
-            if (bufOfs >= blockSize) {
-                // compress completed block now
-                implCompress(buffer, 0);
-                bufOfs = 0;
+    public void update(byte[] in, int inOff, int len) {
+        len = Math.max(0, len);
+
+        //
+        // fill the current word
+        //
+        int i = 0;
+        if (xBufOff != 0) {
+            while (i < len) {
+                xBuf[xBufOff++] = in[inOff + i++];
+                if (xBufOff == 4) {
+                    processWord(xBuf, 0);
+                    xBufOff = 0;
+                    break;
+                }
             }
         }
-        // compress complete blocks
-        while (len >= blockSize) {
-            implCompress(b, ofs);
-            len -= blockSize;
-            ofs += blockSize;
+
+        //
+        // process whole words.
+        //
+        int limit = ((len - i) & ~3) + i;
+        for (; i < limit; i += 4) {
+            processWord(in, inOff + i);
         }
-        // copy remainder to buffer
-        if (len > 0) {
-            System.arraycopy(b, ofs, buffer, 0, len);
-            bufOfs = len;
+
+        //
+        // load in the remainder.
+        //
+        while (i < len) {
+            xBuf[xBufOff++] = in[inOff + i++];
         }
-    }
 
-    private static int FF(int a, int b, int c, int d, int x, int s) {
-        a += ((b & c) | ((~b) & d)) + x;
-        return ((a << s) | (a >>> (32 - s)));
+        byteCount += len;
     }
 
-    private static int GG(int a, int b, int c, int d, int x, int s) {
-        a += ((b & c) | (b & d) | (c & d)) + x + 0x5a827999;
-        return ((a << s) | (a >>> (32 - s)));
+    public void finish() {
+        long bitLength = (byteCount << 3);
+
+        //
+        // add the pad bytes.
+        //
+        update((byte) 128);
+
+        while (xBufOff != 0) {
+            update((byte) 0);
+        }
+
+        processLength(bitLength);
+
+        processBlock();
     }
 
-    private static int HH(int a, int b, int c, int d, int x, int s) {
-        a += ((b ^ c) ^ d) + x + 0x6ed9eba1;
-        return ((a << s) | (a >>> (32 - s)));
+    public int doFinal(byte[] out, int outOff) {
+        finish();
+
+        unpackWord(H1, out, outOff);
+        unpackWord(H2, out, outOff + 4);
+        unpackWord(H3, out, outOff + 8);
+        unpackWord(H4, out, outOff + 12);
+
+        reset();
+
+        return DIGEST_LENGTH;
     }
 
     /**
-     * This is where the functions come together as the generic MD4 transformation operation. It consumes 64 bytes from
-     * the buffer, beginning at the specified offset.
+     * reset the chaining variables to the IV values.
      */
-    private void implCompress(byte[] buf, int ofs) {
-        /**
-         * the following replaces:
-         * 
-         * <pre>
-         * b2iLittle64(buf, ofs, x);
-         * </pre>
-         */
-        for (int xfs = 0; xfs < x.length; xfs++) {
-            x[xfs] = (buf[ofs] & 0xff) | ((buf[ofs + 1] & 0xff) << 8) | ((buf[ofs + 2] & 0xff) << 16)
-                    | ((buf[ofs + 3] & 0xff) << 24);
-            ofs += 4;
+    public void reset() {
+        byteCount = 0;
+
+        xBufOff = 0;
+        for (int i = 0; i < xBuf.length; i++) {
+            xBuf[i] = 0;
         }
 
-        int a = state[0];
-        int b = state[1];
-        int c = state[2];
-        int d = state[3];
-
-        /* Round 1 */
-        a = FF(a, b, c, d, x[0], S11); /* 1 */
-        d = FF(d, a, b, c, x[1], S12); /* 2 */
-        c = FF(c, d, a, b, x[2], S13); /* 3 */
-        b = FF(b, c, d, a, x[3], S14); /* 4 */
-        a = FF(a, b, c, d, x[4], S11); /* 5 */
-        d = FF(d, a, b, c, x[5], S12); /* 6 */
-        c = FF(c, d, a, b, x[6], S13); /* 7 */
-        b = FF(b, c, d, a, x[7], S14); /* 8 */
-        a = FF(a, b, c, d, x[8], S11); /* 9 */
-        d = FF(d, a, b, c, x[9], S12); /* 10 */
-        c = FF(c, d, a, b, x[10], S13); /* 11 */
-        b = FF(b, c, d, a, x[11], S14); /* 12 */
-        a = FF(a, b, c, d, x[12], S11); /* 13 */
-        d = FF(d, a, b, c, x[13], S12); /* 14 */
-        c = FF(c, d, a, b, x[14], S13); /* 15 */
-        b = FF(b, c, d, a, x[15], S14); /* 16 */
-
-        /* Round 2 */
-        a = GG(a, b, c, d, x[0], S21); /* 17 */
-        d = GG(d, a, b, c, x[4], S22); /* 18 */
-        c = GG(c, d, a, b, x[8], S23); /* 19 */
-        b = GG(b, c, d, a, x[12], S24); /* 20 */
-        a = GG(a, b, c, d, x[1], S21); /* 21 */
-        d = GG(d, a, b, c, x[5], S22); /* 22 */
-        c = GG(c, d, a, b, x[9], S23); /* 23 */
-        b = GG(b, c, d, a, x[13], S24); /* 24 */
-        a = GG(a, b, c, d, x[2], S21); /* 25 */
-        d = GG(d, a, b, c, x[6], S22); /* 26 */
-        c = GG(c, d, a, b, x[10], S23); /* 27 */
-        b = GG(b, c, d, a, x[14], S24); /* 28 */
-        a = GG(a, b, c, d, x[3], S21); /* 29 */
-        d = GG(d, a, b, c, x[7], S22); /* 30 */
-        c = GG(c, d, a, b, x[11], S23); /* 31 */
-        b = GG(b, c, d, a, x[15], S24); /* 32 */
-
-        /* Round 3 */
-        a = HH(a, b, c, d, x[0], S31); /* 33 */
-        d = HH(d, a, b, c, x[8], S32); /* 34 */
-        c = HH(c, d, a, b, x[4], S33); /* 35 */
-        b = HH(b, c, d, a, x[12], S34); /* 36 */
-        a = HH(a, b, c, d, x[2], S31); /* 37 */
-        d = HH(d, a, b, c, x[10], S32); /* 38 */
-        c = HH(c, d, a, b, x[6], S33); /* 39 */
-        b = HH(b, c, d, a, x[14], S34); /* 40 */
-        a = HH(a, b, c, d, x[1], S31); /* 41 */
-        d = HH(d, a, b, c, x[9], S32); /* 42 */
-        c = HH(c, d, a, b, x[5], S33); /* 43 */
-        b = HH(b, c, d, a, x[13], S34); /* 44 */
-        a = HH(a, b, c, d, x[3], S31); /* 45 */
-        d = HH(d, a, b, c, x[11], S32); /* 46 */
-        c = HH(c, d, a, b, x[7], S33); /* 47 */
-        b = HH(b, c, d, a, x[15], S34); /* 48 */
-
-        state[0] += a;
-        state[1] += b;
-        state[2] += c;
-        state[3] += d;
+        H1 = 0x67452301;
+        H2 = 0xefcdab89;
+        H3 = 0x98badcfe;
+        H4 = 0x10325476;
+
+        xOff = 0;
+
+        for (int i = 0; i != X.length; i++) {
+            X[i] = 0;
+        }
+    }
+
+    //
+    // round 1 left rotates
+    //
+    private static final int S11 = 3;
+    private static final int S12 = 7;
+    private static final int S13 = 11;
+    private static final int S14 = 19;
+
+    //
+    // round 2 left rotates
+    //
+    private static final int S21 = 3;
+    private static final int S22 = 5;
+    private static final int S23 = 9;
+    private static final int S24 = 13;
+
+    //
+    // round 3 left rotates
+    //
+    private static final int S31 = 3;
+    private static final int S32 = 9;
+    private static final int S33 = 11;
+    private static final int S34 = 15;
+
+    /*
+     * rotate int x left n bits.
+     */
+    private int rotateLeft(int x, int n) {
+        return (x << n) | (x >>> (32 - n));
+    }
+
+    /*
+     * F, G, H and I are the basic MD4 functions.
+     */
+    private int F(int u, int v, int w) {
+        return (u & v) | (~u & w);
+    }
+
+    private int G(int u, int v, int w) {
+        return (u & v) | (u & w) | (v & w);
+    }
+
+    private int H(int u, int v, int w) {
+        return u ^ v ^ w;
+    }
+
+    protected void processBlock() {
+        int a = H1;
+        int b = H2;
+        int c = H3;
+        int d = H4;
+
+        //
+        // Round 1 - F cycle, 16 times.
+        //
+        a = rotateLeft(a + F(b, c, d) + X[0], S11);
+        d = rotateLeft(d + F(a, b, c) + X[1], S12);
+        c = rotateLeft(c + F(d, a, b) + X[2], S13);
+        b = rotateLeft(b + F(c, d, a) + X[3], S14);
+        a = rotateLeft(a + F(b, c, d) + X[4], S11);
+        d = rotateLeft(d + F(a, b, c) + X[5], S12);
+        c = rotateLeft(c + F(d, a, b) + X[6], S13);
+        b = rotateLeft(b + F(c, d, a) + X[7], S14);
+        a = rotateLeft(a + F(b, c, d) + X[8], S11);
+        d = rotateLeft(d + F(a, b, c) + X[9], S12);
+        c = rotateLeft(c + F(d, a, b) + X[10], S13);
+        b = rotateLeft(b + F(c, d, a) + X[11], S14);
+        a = rotateLeft(a + F(b, c, d) + X[12], S11);
+        d = rotateLeft(d + F(a, b, c) + X[13], S12);
+        c = rotateLeft(c + F(d, a, b) + X[14], S13);
+        b = rotateLeft(b + F(c, d, a) + X[15], S14);
+
+        //
+        // Round 2 - G cycle, 16 times.
+        //
+        a = rotateLeft(a + G(b, c, d) + X[0] + 0x5a827999, S21);
+        d = rotateLeft(d + G(a, b, c) + X[4] + 0x5a827999, S22);
+        c = rotateLeft(c + G(d, a, b) + X[8] + 0x5a827999, S23);
+        b = rotateLeft(b + G(c, d, a) + X[12] + 0x5a827999, S24);
+        a = rotateLeft(a + G(b, c, d) + X[1] + 0x5a827999, S21);
+        d = rotateLeft(d + G(a, b, c) + X[5] + 0x5a827999, S22);
+        c = rotateLeft(c + G(d, a, b) + X[9] + 0x5a827999, S23);
+        b = rotateLeft(b + G(c, d, a) + X[13] + 0x5a827999, S24);
+        a = rotateLeft(a + G(b, c, d) + X[2] + 0x5a827999, S21);
+        d = rotateLeft(d + G(a, b, c) + X[6] + 0x5a827999, S22);
+        c = rotateLeft(c + G(d, a, b) + X[10] + 0x5a827999, S23);
+        b = rotateLeft(b + G(c, d, a) + X[14] + 0x5a827999, S24);
+        a = rotateLeft(a + G(b, c, d) + X[3] + 0x5a827999, S21);
+        d = rotateLeft(d + G(a, b, c) + X[7] + 0x5a827999, S22);
+        c = rotateLeft(c + G(d, a, b) + X[11] + 0x5a827999, S23);
+        b = rotateLeft(b + G(c, d, a) + X[15] + 0x5a827999, S24);
+
+        //
+        // Round 3 - H cycle, 16 times.
+        //
+        a = rotateLeft(a + H(b, c, d) + X[0] + 0x6ed9eba1, S31);
+        d = rotateLeft(d + H(a, b, c) + X[8] + 0x6ed9eba1, S32);
+        c = rotateLeft(c + H(d, a, b) + X[4] + 0x6ed9eba1, S33);
+        b = rotateLeft(b + H(c, d, a) + X[12] + 0x6ed9eba1, S34);
+        a = rotateLeft(a + H(b, c, d) + X[2] + 0x6ed9eba1, S31);
+        d = rotateLeft(d + H(a, b, c) + X[10] + 0x6ed9eba1, S32);
+        c = rotateLeft(c + H(d, a, b) + X[6] + 0x6ed9eba1, S33);
+        b = rotateLeft(b + H(c, d, a) + X[14] + 0x6ed9eba1, S34);
+        a = rotateLeft(a + H(b, c, d) + X[1] + 0x6ed9eba1, S31);
+        d = rotateLeft(d + H(a, b, c) + X[9] + 0x6ed9eba1, S32);
+        c = rotateLeft(c + H(d, a, b) + X[5] + 0x6ed9eba1, S33);
+        b = rotateLeft(b + H(c, d, a) + X[13] + 0x6ed9eba1, S34);
+        a = rotateLeft(a + H(b, c, d) + X[3] + 0x6ed9eba1, S31);
+        d = rotateLeft(d + H(a, b, c) + X[11] + 0x6ed9eba1, S32);
+        c = rotateLeft(c + H(d, a, b) + X[7] + 0x6ed9eba1, S33);
+        b = rotateLeft(b + H(c, d, a) + X[15] + 0x6ed9eba1, S34);
+
+        H1 += a;
+        H2 += b;
+        H3 += c;
+        H4 += d;
+
+        //
+        // reset the offset and clean out the word buffer.
+        //
+        xOff = 0;
+        for (int i = 0; i != X.length; i++) {
+            X[i] = 0;
+        }
     }
 }


Test patch:
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/AlwaysEncrypted/AESetup.java b/src/test/java/com/microsoft/sqlserver/jdbc/AlwaysEncrypted/AESetup.java
index 49a435a6..de367d75 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/AlwaysEncrypted/AESetup.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/AlwaysEncrypted/AESetup.java
@@ -61,7 +61,6 @@ public class AESetup extends AbstractTest {
     static String cekJks = Constants.CEK_NAME + "_JKS";
     static String cekWin = Constants.CEK_NAME + "_WIN";
     static String cekAkv = Constants.CEK_NAME + "_AKV";
-
     static SQLServerStatementColumnEncryptionSetting stmtColEncSetting = null;
 
     static String AETestConnectionString;
@@ -218,17 +217,17 @@ public class AESetup extends AbstractTest {
 
             setAEConnectionString(serverName, url, protocol);
 
-            createCMK(cmkJks, Constants.JAVA_KEY_STORE_NAME, javaKeyAliases, Constants.CMK_SIGNATURE);
-            createCEK(cmkJks, cekJks, jksProvider);
+            createCMK(AETestConnectionString, cmkJks, Constants.JAVA_KEY_STORE_NAME, javaKeyAliases, Constants.CMK_SIGNATURE);
+            createCEK(AETestConnectionString, cmkJks, cekJks, jksProvider);
 
             if (null != keyIDs && !keyIDs[0].isEmpty()) {
-                createCMK(cmkAkv, Constants.AZURE_KEY_VAULT_NAME, keyIDs[0], Constants.CMK_SIGNATURE_AKV);
-                createCEK(cmkAkv, cekAkv, akvProvider);
+                createCMK(AETestConnectionString, cmkAkv, Constants.AZURE_KEY_VAULT_NAME, keyIDs[0], Constants.CMK_SIGNATURE_AKV);
+                createCEK(AETestConnectionString, cmkAkv, cekAkv, akvProvider);
             }
 
             if (null != windowsKeyPath) {
-                createCMK(cmkWin, Constants.WINDOWS_KEY_STORE_NAME, windowsKeyPath, Constants.CMK_SIGNATURE);
-                createCEK(cmkWin, cekWin, null);
+                createCMK(AETestConnectionString, cmkWin, Constants.WINDOWS_KEY_STORE_NAME, windowsKeyPath, Constants.CMK_SIGNATURE);
+                createCEK(AETestConnectionString, cmkWin, cekWin, null);
             }
         }
     }
@@ -505,10 +504,10 @@ public class AESetup extends AbstractTest {
      * @param keyPath
      * @throws SQLException
      */
-    private static void createCMK(String cmkName, String keyStoreName, String keyPath,
+    protected static void createCMK(String connectionString, String cmkName, String keyStoreName, String keyPath,
             String signature) throws SQLException {
         try (SQLServerConnection con = (SQLServerConnection) PrepUtil
-                .getConnection(AETestConnectionString + ";sendTimeAsDateTime=false", AEInfo);
+                .getConnection(connectionString + ";sendTimeAsDateTime=false", AEInfo);
                 SQLServerStatement stmt = (SQLServerStatement) con.createStatement()) {
             String sql = " if not exists (SELECT name from sys.column_master_keys where name='" + cmkName + "')"
                     + " begin" + " CREATE COLUMN MASTER KEY " + cmkName + " WITH (KEY_STORE_PROVIDER_NAME = '"
@@ -525,10 +524,10 @@ public class AESetup extends AbstractTest {
      * @param certStore
      * @throws SQLException
      */
-    private static void createCEK(String cmkName, String cekName,
+    protected static void createCEK(String connectionString, String cmkName, String cekName,
             SQLServerColumnEncryptionKeyStoreProvider storeProvider) throws SQLException {
         try (SQLServerConnection con = (SQLServerConnection) PrepUtil
-                .getConnection(AETestConnectionString + ";sendTimeAsDateTime=false", AEInfo);
+                .getConnection(connectionString + ";sendTimeAsDateTime=false", AEInfo);
                 SQLServerStatement stmt = (SQLServerStatement) con.createStatement()) {
             byte[] valuesDefault = Constants.CEK_STRING.getBytes();
             String encryptedValue;
@@ -1923,7 +1922,7 @@ public class AESetup extends AbstractTest {
      * 
      * @throws SQLException
      */
-    private static void dropCEK(String cekName, Statement stmt) throws SQLException {
+    protected static void dropCEK(String cekName, Statement stmt) throws SQLException {
         String cekSql = " if exists (SELECT name from sys.column_encryption_keys where name='" + cekName + "')"
                 + " begin" + " drop column encryption key " + cekName + " end";
         stmt.execute(cekSql);
@@ -1934,12 +1933,33 @@ public class AESetup extends AbstractTest {
      * 
      * @throws SQLException
      */
-    private static void dropCMK(String cmkName, Statement stmt) throws SQLException {
+    protected static void dropCMK(String cmkName, Statement stmt) throws SQLException {
         String cekSql = " if exists (SELECT name from sys.column_master_keys where name='" + cmkName + "')" + " begin"
                 + " drop column master key " + cmkName + " end";
         stmt.execute(cekSql);
     }
 
+    protected static void dropObject(String connString, String objectType, String objectName) {
+        try (SQLServerConnection con = (SQLServerConnection) PrepUtil.getConnection(connString, AEInfo);
+                SQLServerStatement stmt = (SQLServerStatement) con.createStatement()) {
+            switch(objectType) {
+                case "TABLE":
+                    TestUtils.dropTableIfExists(objectName, stmt);
+                    break;
+                case "CEK":
+                    dropCEK(objectName, stmt);
+                    break;
+                case "CMK":
+                    dropCMK(objectName, stmt);
+                    break;
+                default:
+                    break;
+            }
+        } catch(Exception ex) {
+            fail(ex.getMessage());
+        }
+    }
+
     /**
      * Alter Column encryption on deterministic columns to randomized - this will trigger enclave to re-encrypt
      * 
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/AlwaysEncrypted/MultiUserAKVTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/AlwaysEncrypted/MultiUserAKVTest.java
new file mode 100644
index 00000000..bfd7453a
--- /dev/null
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/AlwaysEncrypted/MultiUserAKVTest.java
@@ -0,0 +1,637 @@
+/*
+ * Microsoft JDBC Driver for SQL Server Copyright(c) Microsoft Corporation All rights reserved. This program is made
+ * available under the terms of the MIT License. See the LICENSE file in the project root for more information.
+ */
+package com.microsoft.sqlserver.jdbc.AlwaysEncrypted;
+
+import com.microsoft.sqlserver.testframework.DummyKeyStoreProvider;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.security.MessageDigest;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.OutputStream;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.text.MessageFormat;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.platform.runner.JUnitPlatform;
+import org.junit.runner.RunWith;
+
+import com.microsoft.sqlserver.jdbc.RandomUtil;
+import com.microsoft.sqlserver.jdbc.SQLServerColumnEncryptionAzureKeyVaultProvider;
+import com.microsoft.sqlserver.jdbc.SQLServerColumnEncryptionKeyStoreProvider;
+import com.microsoft.sqlserver.jdbc.SQLServerConnection;
+import com.microsoft.sqlserver.jdbc.SQLServerException;
+import com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement;
+import com.microsoft.sqlserver.jdbc.SQLServerStatement;
+import com.microsoft.sqlserver.jdbc.SQLServerStatementColumnEncryptionSetting;
+import com.microsoft.sqlserver.jdbc.TestResource;
+import com.microsoft.sqlserver.jdbc.TestUtils;
+import com.microsoft.sqlserver.testframework.AbstractSQLGenerator;
+import com.microsoft.sqlserver.testframework.Constants;
+import com.microsoft.sqlserver.testframework.PrepUtil;
+
+
+/**
+ * Test multi-user Azure Key Store provider.
+ */
+@RunWith(JUnitPlatform.class)
+@Tag(Constants.xSQLv12)
+@Tag(Constants.xAzureSQLDW)
+@Tag(Constants.xAzureSQLDB)
+public class MultiUserAKVTest extends AESetup {
+
+    private static Map<String, SQLServerColumnEncryptionKeyStoreProvider> requiredKeyStoreProvider = new HashMap<>();
+    private static Map<String, SQLServerColumnEncryptionKeyStoreProvider> notRequiredKeyStoreProvider = new HashMap<>();
+
+    private static final String notRequiredProviderName = "UNWANTED_DUMMY_PROVIDER";
+    private static final String requiredProviderName = "DUMMY_PROVIDER";
+    private static final String cekCacheSizeGetterName = "getColumnEncryptionKeyCacheSize";
+    private static final String cmkMetadataCacheSizeGetterName = "getCmkMetadataSignatureVerificationCacheSize";
+    private static final String cekCacheName = "columnEncryptionKeyCache";
+
+    private static final String customProviderTableName = TestUtils
+            .escapeSingleQuotes(AbstractSQLGenerator.escapeIdentifier(RandomUtil.getIdentifier("JDBCCustomProvider")));
+
+    private static final String dummyProviderTableName = TestUtils
+            .escapeSingleQuotes(AbstractSQLGenerator.escapeIdentifier(RandomUtil.getIdentifier("JDBCDummyProvider")));
+
+    private static String cmkDummy = Constants.CMK_NAME + "_DUMMY"; // For dummyKeyStoreProvider
+    private static String cekDummy = Constants.CEK_NAME + "_DUMMY"; // For dummyKeyStoreProvider
+
+    private static boolean isMasterKeyPathSetup = false;
+
+    @BeforeAll
+    public static void testSetup() throws Exception {
+        requiredKeyStoreProvider.put(requiredProviderName, new DummyKeyStoreProvider());
+        notRequiredKeyStoreProvider.put(notRequiredProviderName, new DummyKeyStoreProvider());
+
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+
+        isMasterKeyPathSetup = !(null == keyIDs[0] || keyIDs[0].trim().isEmpty());
+    }
+
+    @AfterAll
+    public static void testCleanUp() throws Exception {
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> tempMap = new HashMap<String, SQLServerColumnEncryptionKeyStoreProvider>();
+
+        if (null != jksProvider) {
+            tempMap.put(Constants.CUSTOM_KEYSTORE_NAME, jksProvider);
+        }
+
+        if (null != akvProvider && null != applicationClientID && null != applicationKey) {
+            tempMap.put(Constants.AZURE_KEY_VAULT_NAME, akvProvider);
+        }
+
+        if (tempMap.size() > 0) {
+            SQLServerConnection.registerColumnEncryptionKeyStoreProviders(tempMap);
+        }
+    }
+
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    public void decryptedCekIsCachedDuringDecryption() throws Exception {
+        SQLServerColumnEncryptionAzureKeyVaultProvider provider = createAKVProvider();
+
+        if (null == provider) {
+            fail(TestResource.getResource("R_AKVProviderNull"));
+        }
+
+        if (!isMasterKeyPathSetup) {
+            Object[] msgArg = {"master key path"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        byte[] plaintextKey1 = {1, 2, 3};
+        byte[] plaintextKey2 = {1, 2, 3};
+        byte[] plaintextKey3 = {0, 1, 2, 3};
+
+        byte[] encryptedKey1 = provider.encryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, plaintextKey1);
+        byte[] encryptedKey2 = provider.encryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, plaintextKey2);
+        byte[] encryptedKey3 = provider.encryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, plaintextKey3);
+
+        byte[] decryptedKey1 = provider.decryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, encryptedKey1);
+        assertEquals(1, getCacheSize(cekCacheSizeGetterName, provider));
+        assertArrayEquals(plaintextKey1, decryptedKey1);
+
+        // this should not create a new entry in cek cache
+        decryptedKey1 = provider.decryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, encryptedKey1);
+        assertEquals(1, getCacheSize(cekCacheSizeGetterName, provider));
+        assertArrayEquals(plaintextKey1, decryptedKey1);
+
+        byte[] decryptedKey2 = provider.decryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, encryptedKey2);
+        assertEquals(2, getCacheSize(cekCacheSizeGetterName, provider));
+        assertArrayEquals(plaintextKey2, decryptedKey2);
+
+        byte[] decryptedKey3 = provider.decryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, encryptedKey3);
+        assertEquals(3, getCacheSize(cekCacheSizeGetterName, provider));
+        assertArrayEquals(plaintextKey3, decryptedKey3);
+
+    }
+
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    public void signatureVerificationResultIsCachedDuringVerification() throws Exception {
+        SQLServerColumnEncryptionAzureKeyVaultProvider provider = createAKVProvider();
+
+        if (provider == null) {
+            fail(TestResource.getResource("R_AKVProviderNull"));
+        }
+
+        if (!isMasterKeyPathSetup) {
+            Object[] msgArg = {"master key path"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        byte[] signature1 = signColumnMasterKeyMetadata(provider, keyIDs[0], true);
+        byte[] signature2 = signColumnMasterKeyMetadata(provider, keyIDs[0], true);
+        byte[] signatureWithoutEnclave = signColumnMasterKeyMetadata(provider, keyIDs[0], false);
+
+        assertTrue(provider.verifyColumnMasterKeyMetadata(keyIDs[0], true, signature1));
+        assertEquals(1, getCacheSize(cmkMetadataCacheSizeGetterName, provider));
+
+        // this should not create a new entry in cmk metadata signature verification cache
+        assertTrue(provider.verifyColumnMasterKeyMetadata(keyIDs[0], true, signature1));
+        assertEquals(1, getCacheSize(cmkMetadataCacheSizeGetterName, provider));
+
+        assertTrue(provider.verifyColumnMasterKeyMetadata(keyIDs[0], true, signature2));
+        assertEquals(1, getCacheSize(cmkMetadataCacheSizeGetterName, provider));
+
+        assertFalse(provider.verifyColumnMasterKeyMetadata(keyIDs[0], false, signatureWithoutEnclave));
+        assertEquals(1, getCacheSize(cmkMetadataCacheSizeGetterName, provider));
+    }
+
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    public void cekCacheEntryIsEvictedAfterTtlExpires() throws Exception {
+        SQLServerColumnEncryptionAzureKeyVaultProvider provider = createAKVProvider();
+
+        if (provider == null) {
+            fail(TestResource.getResource("R_AKVProviderNull"));
+        }
+
+        if (!isMasterKeyPathSetup) {
+            Object[] msgArg = {"master key path"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        provider.setColumnEncryptionCacheTtl(Duration.ofSeconds(10));
+        byte[] plaintextKey = {1, 2, 3};
+        byte[] encryptedKey = provider.encryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, plaintextKey);
+
+        provider.decryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, encryptedKey);
+        assertTrue(cekCacheContainsKey(encryptedKey, provider));
+        assertEquals(1, getCacheSize(cekCacheSizeGetterName, provider));
+
+        Thread.sleep(TimeUnit.SECONDS.toMillis(11));
+        assertFalse(cekCacheContainsKey(encryptedKey, provider));
+        assertEquals(0, getCacheSize(cekCacheSizeGetterName, provider));
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    public void cekCacheShouldBeDisabledWhenAkvProviderIsRegisteredGlobally() throws Exception {
+        SQLServerColumnEncryptionAzureKeyVaultProvider provider = createAKVProvider();
+
+        if (provider == null) {
+            fail(TestResource.getResource("R_AKVProviderNull"));
+        }
+
+        if (!isMasterKeyPathSetup) {
+            Object[] msgArg = {"master key path"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providerMap = new HashMap<String, SQLServerColumnEncryptionKeyStoreProvider>();
+        providerMap.put(Constants.AZURE_KEY_VAULT_NAME, provider);
+        SQLServerConnection.registerColumnEncryptionKeyStoreProviders(providerMap);
+
+        try (SQLServerConnection con = (SQLServerConnection) PrepUtil.getConnection(AETestConnectionString, AEInfo)) {
+            Field globalCustomCacheField = SQLServerConnection.class
+                    .getDeclaredField("globalCustomColumnEncryptionKeyStoreProviders");
+            globalCustomCacheField.setAccessible(true);
+
+            Map<String, SQLServerColumnEncryptionKeyStoreProvider> globalCacheFieldValue = (Map<String, SQLServerColumnEncryptionKeyStoreProvider>) globalCustomCacheField
+                    .get(con);
+
+            Method method = globalCacheFieldValue.getClass().getDeclaredMethod("get", Object.class);
+            SQLServerColumnEncryptionAzureKeyVaultProvider providerInGlobalCache = (SQLServerColumnEncryptionAzureKeyVaultProvider) method
+                    .invoke(globalCacheFieldValue, Constants.AZURE_KEY_VAULT_NAME);
+
+            byte[] plaintextKey = {1, 2, 3};
+            byte[] encryptedKey = providerInGlobalCache.encryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM,
+                    plaintextKey);
+
+            providerInGlobalCache.decryptColumnEncryptionKey(keyIDs[0], Constants.CEK_ALGORITHM, encryptedKey);
+            assertEquals(0, getCacheSize(cekCacheSizeGetterName, providerInGlobalCache));
+        } catch (SQLException e) {
+            fail(e.getMessage());
+        } finally {
+            // clean up
+            SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        }
+    }
+
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    public void testLocalCekCacheIsScopedToProvider() throws Exception {
+        SQLServerColumnEncryptionAzureKeyVaultProvider provider = createAKVProvider();
+
+        if (provider == null) {
+            fail(TestResource.getResource("R_AKVProviderNull"));
+        }
+
+        if (!isMasterKeyPathSetup) {
+            Object[] msgArg = {"master key path"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providerMap = new HashMap<String, SQLServerColumnEncryptionKeyStoreProvider>();
+        providerMap.put(Constants.AZURE_KEY_VAULT_NAME, akvProvider);
+        SQLServerConnection.registerColumnEncryptionKeyStoreProviders(providerMap);
+
+        int customerId = 10;
+        String customerName = "Microsoft";
+        createTableForCustomProvider(AETestConnectionString, customProviderTableName, cekAkv);
+        insertData(customProviderTableName, customerId, customerName);
+
+        String sql = "SELECT CustomerId, CustomerName FROM " + customProviderTableName + " WHERE CustomerId = ?";
+
+        try (SQLServerConnection con = PrepUtil.getConnection(AETestConnectionString, AEInfo);
+                SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sql,
+                        SQLServerStatementColumnEncryptionSetting.Enabled)) {
+
+            pstmt.setInt(1, customerId);
+
+            try (ResultSet rs = pstmt.executeQuery()) {
+                while (rs.next()) {
+                    int intValue = rs.getInt(1);
+                    String strValue = rs.getString(2);
+                    assertTrue((customerId == intValue) && strValue.equalsIgnoreCase(customerName));
+                }
+            }
+
+            // Clean up global custom providers
+            SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+
+            // Register key store provider on statement level
+            providerMap.put(Constants.AZURE_KEY_VAULT_NAME, provider);
+            pstmt.registerColumnEncryptionKeyStoreProvidersOnStatement(providerMap);
+
+            // Execute a query using provider from statement-level cache. this will cache the cek in the local cek cache
+            try (ResultSet rs2 = pstmt.executeQuery()) {
+                while (rs2.next()) {
+                    int intValue = rs2.getInt(1);
+                    String strValue = rs2.getString(2);
+                    assertTrue((customerId == intValue) && strValue.equalsIgnoreCase(customerName));
+                }
+            }
+
+            // Register invalid key store provider on statement level. This will overwrite the previous one.
+            SQLServerColumnEncryptionAzureKeyVaultProvider providerWithBadCred = new SQLServerColumnEncryptionAzureKeyVaultProvider(
+                    "badApplicationID", "badApplicationKey");
+            providerMap.put(Constants.AZURE_KEY_VAULT_NAME, providerWithBadCred);
+            pstmt.registerColumnEncryptionKeyStoreProvidersOnStatement(providerMap);
+
+            // The following query should fail due to an empty cek cache and invalid credentials
+            try (ResultSet rs3 = pstmt.executeQuery()) {
+                while (rs3.next()) {
+                    int intValue = rs3.getInt(1);
+                    String strValue = rs3.getString(2);
+                    assertTrue((customerId == intValue) && strValue.equalsIgnoreCase(customerName));
+                }
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (SQLServerException ex) {
+                assertTrue(ex.getMessage().contains("AADSTS700016"));
+            }
+        } finally {
+            dropObject(AETestConnectionString, "TABLE", customProviderTableName);
+            SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        }
+    }
+
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    public void testConnectionCustomKeyStoreProviderDuringAeQuery() throws Exception {
+        DummyKeyStoreProvider dummyProvider = new DummyKeyStoreProvider();
+
+        if (akvProvider == null) {
+            fail(TestResource.getResource("R_AKVProviderNull"));
+        }
+
+        if (!isMasterKeyPathSetup) {
+            Object[] msgArg = {"master key path"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providerMap = new HashMap<String, SQLServerColumnEncryptionKeyStoreProvider>();
+        providerMap.put(Constants.DUMMY_KEYSTORE_NAME, dummyProvider);
+        SQLServerConnection.registerColumnEncryptionKeyStoreProviders(providerMap);
+
+        // Create cmk and cek for DummyKeyStoreProvider
+        createCMK(AETestConnectionString, cmkDummy, Constants.DUMMY_KEYSTORE_NAME, keyIDs[0],
+                Constants.CMK_SIGNATURE_AKV);
+        createCEK(AETestConnectionString, cmkDummy, cekDummy, akvProvider);
+
+        // Create an empty table for testing
+        createTableForCustomProvider(AETestConnectionString, dummyProviderTableName, cekDummy);
+
+        int customerId = 10;
+        String sql = "SELECT CustomerId, CustomerName FROM " + dummyProviderTableName + " WHERE CustomerId = ?";
+
+        try (SQLServerConnection con = PrepUtil.getConnection(AETestConnectionString, AEInfo)) {
+            try (SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sql,
+                    SQLServerStatementColumnEncryptionSetting.Enabled)) {
+                pstmt.setInt(1, customerId);
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex instanceof UnsupportedOperationException);
+            }
+
+            /*
+             * Register not required provider at connection instance level. It should not fall back to the global cache
+             * so the right provider will not be found.
+             */
+            con.registerColumnEncryptionKeyStoreProvidersOnConnection(notRequiredKeyStoreProvider);
+            try (SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sql,
+                    SQLServerStatementColumnEncryptionSetting.Enabled)) {
+                pstmt.setInt(1, customerId);
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex.getMessage()
+                        .matches(TestUtils.formatErrorMsg("R_UnrecognizedConnectionKeyStoreProviderName")));
+            }
+
+            /*
+             * Required provider in connection instance cache. If the instance cache is not empty, it is always checked
+             * for the provider. If the provider is found, it must have been retrieved from the instance cache and not
+             * the global cache.
+             */
+            con.registerColumnEncryptionKeyStoreProvidersOnConnection(requiredKeyStoreProvider);
+            try (SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sql,
+                    SQLServerStatementColumnEncryptionSetting.Enabled)) {
+                pstmt.setInt(1, customerId);
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex instanceof UnsupportedOperationException);
+            }
+
+            // Not required provider will replace the previous entry so required provider will not be found.
+            con.registerColumnEncryptionKeyStoreProvidersOnConnection(notRequiredKeyStoreProvider);
+            try (SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sql,
+                    SQLServerStatementColumnEncryptionSetting.Enabled)) {
+                pstmt.setInt(1, customerId);
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex.getMessage()
+                        .matches(TestUtils.formatErrorMsg("R_UnrecognizedConnectionKeyStoreProviderName")));
+            }
+        } finally {
+            dropObject(AETestConnectionString, "TABLE", dummyProviderTableName);
+            dropObject(AETestConnectionString, "CEK", cekDummy);
+            dropObject(AETestConnectionString, "CMK", cmkDummy);
+            SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        }
+    }
+
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    public void testStatementCustomKeyStoreProviderDuringAeQuery() throws Exception {
+        DummyKeyStoreProvider dummyProvider = new DummyKeyStoreProvider();
+
+        if (!isMasterKeyPathSetup) {
+            Object[] msgArg = {"master key path"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providerMap = new HashMap<String, SQLServerColumnEncryptionKeyStoreProvider>();
+        providerMap.put(Constants.DUMMY_KEYSTORE_NAME, dummyProvider);
+        SQLServerConnection.registerColumnEncryptionKeyStoreProviders(providerMap);
+
+        // Create an empty table for testing
+        createCMK(AETestConnectionString, cmkDummy, Constants.DUMMY_KEYSTORE_NAME, keyIDs[0],
+                Constants.CMK_SIGNATURE_AKV);
+        createCEK(AETestConnectionString, cmkDummy, cekDummy, akvProvider);
+
+        createTableForCustomProvider(AETestConnectionString, customProviderTableName, cekDummy);
+
+        int customerId = 10;
+        String sql = "SELECT CustomerId, CustomerName FROM " + customProviderTableName + " WHERE CustomerId = ?";
+
+        try (SQLServerConnection con = PrepUtil.getConnection(AETestConnectionString, AEInfo);
+                SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sql,
+                        SQLServerStatementColumnEncryptionSetting.Enabled)) {
+            pstmt.setInt(1, customerId);
+
+            /*
+             * DummyProvider in global cache will be used. Provider will be found but it will throw when its methods are
+             * called.
+             */
+            try {
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex instanceof UnsupportedOperationException);
+            }
+
+            // Required provider will be found in statement instance level.
+            pstmt.registerColumnEncryptionKeyStoreProvidersOnStatement(requiredKeyStoreProvider);
+            try {
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex instanceof UnsupportedOperationException);
+            }
+
+            /*
+             * Register not required provider at statement instance level. It should not fall back to the global cache
+             * so the right provider will not be found.
+             */
+            pstmt.registerColumnEncryptionKeyStoreProvidersOnStatement(notRequiredKeyStoreProvider);
+            try {
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex.getMessage()
+                        .matches(TestUtils.formatErrorMsg("R_UnrecognizedStatementKeyStoreProviderName")));
+            }
+
+            /*
+             * Register required provider at connection level but keep not required provider at statement level. This
+             * should not fall back to connection level or global level.
+             */
+            con.registerColumnEncryptionKeyStoreProvidersOnConnection(requiredKeyStoreProvider);
+            try {
+                pstmt.executeQuery();
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (Exception ex) {
+                assertTrue(ex.getMessage()
+                        .matches(TestUtils.formatErrorMsg("R_UnrecognizedStatementKeyStoreProviderName")));
+            }
+
+            /*
+             * The new statement instance should have an empty cache and query will fall back to connection level which
+             * contains the required provider
+             */
+            try (SQLServerPreparedStatement pstmt2 = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sql,
+                    SQLServerStatementColumnEncryptionSetting.Enabled)) {
+                pstmt2.setInt(1, customerId);
+
+                try {
+                    pstmt2.executeQuery();
+                    fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+                } catch (Exception ex) {
+                    assertTrue(ex instanceof UnsupportedOperationException);
+                }
+            }
+        } finally {
+            dropObject(AETestConnectionString, "TABLE", customProviderTableName);
+            dropObject(AETestConnectionString, "CEK", cekDummy);
+            dropObject(AETestConnectionString, "CMK", cmkDummy);
+
+            SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+        }
+    }
+
+    private void insertData(String tableName, int customId, String customName) {
+        String sqlQuery = "INSERT INTO " + tableName + " VALUES ( ?, ? )";
+
+        try (SQLServerConnection con = PrepUtil.getConnection(AETestConnectionString, AEInfo);
+                SQLServerPreparedStatement pstmt = (SQLServerPreparedStatement) TestUtils.getPreparedStmt(con, sqlQuery,
+                        SQLServerStatementColumnEncryptionSetting.Enabled)) {
+            pstmt.setInt(1, customId);
+            pstmt.setString(2, customName);
+            pstmt.executeUpdate();
+        } catch (SQLException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    private void createTableForCustomProvider(String connString, String tableName, String cekName) {
+        String sqlQuery = "CREATE TABLE " + tableName + " ("
+                + " [CustomerId] [int] ENCRYPTED WITH (COLUMN_ENCRYPTION_KEY = [" + cekName
+                + "], ENCRYPTION_TYPE = DETERMINISTIC, ALGORITHM = 'AEAD_AES_256_CBC_HMAC_SHA_256'), "
+                + " [CustomerName] [varchar](50) COLLATE Latin1_General_BIN2 ENCRYPTED WITH (COLUMN_ENCRYPTION_KEY = ["
+                + cekName + "], ENCRYPTION_TYPE = DETERMINISTIC, ALGORITHM = 'AEAD_AES_256_CBC_HMAC_SHA_256') " + ")";
+
+        try (SQLServerConnection con = PrepUtil.getConnection(connString, AEInfo);
+                SQLServerStatement stmt = (SQLServerStatement) con.createStatement()) {
+            TestUtils.dropTableIfExists(tableName, stmt);
+            stmt.execute(sqlQuery);
+        } catch (SQLException e) {
+            fail(e.getMessage());
+        }
+    }
+
+    private boolean cekCacheContainsKey(byte[] encryptedKey,
+            SQLServerColumnEncryptionAzureKeyVaultProvider provider) throws Exception {
+        assertFalse(null == encryptedKey || 0 == encryptedKey.length);
+
+        String encryptedCEKHexString = TestUtils.byteToHexDisplayString(encryptedKey);
+
+        Field cekCacheField = provider.getClass().getDeclaredField(cekCacheName);
+        cekCacheField.setAccessible(true);
+
+        Object fieldValue = cekCacheField.get(provider);
+        Method method = fieldValue.getClass().getDeclaredMethod("contains", Object.class);
+        method.setAccessible(true);
+
+        boolean result = (boolean) method.invoke(fieldValue, encryptedCEKHexString);
+
+        return result;
+    }
+
+    private byte[] signColumnMasterKeyMetadata(SQLServerColumnEncryptionAzureKeyVaultProvider provider,
+            String masterKeyPath, boolean allowEnclaveComputations) throws Exception {
+
+        MessageDigest md = MessageDigest.getInstance("SHA-256");
+        md.update(Constants.AZURE_KEY_VAULT_NAME.toLowerCase().getBytes(java.nio.charset.StandardCharsets.UTF_16LE));
+        md.update(masterKeyPath.toLowerCase().getBytes(java.nio.charset.StandardCharsets.UTF_16LE));
+        md.update(Boolean.toString(allowEnclaveComputations).getBytes(java.nio.charset.StandardCharsets.UTF_16LE));
+
+        byte[] dataToSign = md.digest();
+        if (null == dataToSign) {
+            fail("data to sign is null or empty.");
+            Object[] msgArg = {"dataToSign"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        assertTrue(dataToSign.length > 0);
+
+        Method method = provider.getClass().getDeclaredMethod("AzureKeyVaultSignHashedData", byte[].class,
+                String.class);
+        method.setAccessible(true);
+
+        byte[] signature = (byte[]) method.invoke(provider, dataToSign, masterKeyPath);
+
+        if (null == signature || 0 == signature.length) {
+            Object[] msgArg = {"Signature of column master key metadata"};
+            fail((new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"))).format(msgArg));
+        }
+
+        return signature;
+    }
+
+    private int getCacheSize(String methodName,
+            SQLServerColumnEncryptionAzureKeyVaultProvider provider) throws Exception {
+        Method method = provider.getClass().getDeclaredMethod(methodName);
+        method.setAccessible(true);
+
+        return (int) method.invoke(provider);
+    }
+
+    private SQLServerColumnEncryptionAzureKeyVaultProvider createAKVProvider() throws Exception {
+
+        SQLServerColumnEncryptionAzureKeyVaultProvider azureKeyVaultProvider = null;
+
+        if (null != applicationClientID && null != applicationKey) {
+            File file = null;
+            try {
+                file = new File(Constants.MSSQL_JDBC_PROPERTIES);
+                try (OutputStream os = new FileOutputStream(file);) {
+                    Properties props = new Properties();
+                    // Append to the list of hardcoded endpoints
+                    props.setProperty(Constants.AKV_TRUSTED_ENDPOINTS_KEYWORD, ";vault.azure.net");
+                    props.store(os, "");
+                }
+                azureKeyVaultProvider = new SQLServerColumnEncryptionAzureKeyVaultProvider(applicationClientID,
+                        applicationKey);
+
+            } finally {
+                if (null != file) {
+                    file.delete();
+                }
+            }
+        }
+
+        return azureKeyVaultProvider;
+    }
+}
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/EnclavePackageTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/EnclavePackageTest.java
index 64b2c461..40cd4a10 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/EnclavePackageTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/EnclavePackageTest.java
@@ -329,7 +329,7 @@ public class EnclavePackageTest extends AbstractTest {
             trustedKeyPaths.put(serverName, paths);
             SQLServerConnection.setColumnEncryptionTrustedMasterKeyPaths(trustedKeyPaths);
 
-            SQLServerSecurityUtility.verifyColumnMasterKeyMetadata(connection, "My_KEYSTORE", "UnTrustedKeyPath",
+            SQLServerSecurityUtility.verifyColumnMasterKeyMetadata(connection, null, "My_KEYSTORE", "UnTrustedKeyPath",
                     serverName, true, null);
             fail(TestResource.getResource("R_expectedFailPassed"));
         } catch (SQLServerException e) {
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/RegisterKeyStoreProviderTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/RegisterKeyStoreProviderTest.java
new file mode 100644
index 00000000..f640e785
--- /dev/null
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/RegisterKeyStoreProviderTest.java
@@ -0,0 +1,202 @@
+/*
+ * Microsoft JDBC Driver for SQL Server Copyright(c) Microsoft Corporation All rights reserved. This program is made
+ * available under the terms of the MIT License. See the LICENSE file in the project root for more information.
+ */
+package com.microsoft.sqlserver.jdbc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.platform.runner.JUnitPlatform;
+import org.junit.runner.RunWith;
+
+import com.microsoft.sqlserver.testframework.AbstractTest;
+import com.microsoft.sqlserver.testframework.Constants;
+import com.microsoft.sqlserver.testframework.DummyKeyStoreProvider;
+
+
+/**
+ * Test key store provider registration.
+ */
+@RunWith(JUnitPlatform.class)
+public class RegisterKeyStoreProviderTest extends AbstractTest {
+
+    private static final String dummyProviderName1 = "DummyProvider1";
+    private static final String dummyProviderName2 = "DummyProvider2";
+    private static final String dummyProviderName3 = "DummyProvider3";
+    private static final String dummyProviderName4 = "DummyProvider4";
+
+    private static Map<String, SQLServerColumnEncryptionKeyStoreProvider> singleKeyStoreProvider = new HashMap<>();
+    private static Map<String, SQLServerColumnEncryptionKeyStoreProvider> multipleKeyStoreProviders = new HashMap<>();
+    private static Map<String, SQLServerColumnEncryptionKeyStoreProvider> globalKeyStoreProviders = null;
+
+    @BeforeAll
+    public static void testSetup() throws Exception {
+        singleKeyStoreProvider.put(dummyProviderName1, new DummyKeyStoreProvider());
+
+        multipleKeyStoreProviders.put(dummyProviderName2, new DummyKeyStoreProvider());
+        multipleKeyStoreProviders.put(dummyProviderName3, new DummyKeyStoreProvider());
+        multipleKeyStoreProviders.put(dummyProviderName4, new DummyKeyStoreProvider());
+
+        if (null != SQLServerConnection.globalCustomColumnEncryptionKeyStoreProviders) {
+            globalKeyStoreProviders = new HashMap<String, SQLServerColumnEncryptionKeyStoreProvider>(
+                    SQLServerConnection.globalCustomColumnEncryptionKeyStoreProviders);
+        }
+
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+    }
+
+    @Test
+    public void testNullProviderMap() {
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers = null;
+        assertExceptionIsThrownForAllCustomProviderCaches(providers,
+                TestUtils.formatErrorMsg("R_CustomKeyStoreProviderMapNull"));
+    }
+
+    @Test
+    public void testInvalidProviderNameOnGlobal() {
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers = new HashMap<>();
+        providers.put("MSSQL_DUMMY", new DummyKeyStoreProvider());
+
+        assertExceptionIsThrownForGlobalCustomProviderCache(providers,
+                TestUtils.formatErrorMsg("R_InvalidCustomKeyStoreProviderName"));
+    }
+
+    @Test
+    public void testInvalidProviderNameAllLevels() {
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers = new HashMap<>();
+        providers.put(Constants.WINDOWS_KEY_STORE_NAME, new DummyKeyStoreProvider());
+        assertExceptionIsThrownForAllCustomProviderCaches(providers,
+                TestUtils.formatErrorMsg("R_InvalidCustomKeyStoreProviderName"));
+    }
+
+    @Test
+    public void testNullProviderAllLevels() {
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers = new HashMap<>();
+        providers.put("DUMMY_PROVIDER", null);
+        assertExceptionIsThrownForAllCustomProviderCaches(providers,
+                TestUtils.formatErrorMsg("R_CustomKeyStoreProviderValueNull"));
+
+    }
+
+    @Test
+    public void testEmptyProviderNameAllLevels() {
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers = new HashMap<>();
+        providers.put(" ", new DummyKeyStoreProvider());
+        assertExceptionIsThrownForAllCustomProviderCaches(providers,
+                TestUtils.formatErrorMsg("R_EmptyCustomKeyStoreProviderName"));
+    }
+
+    @Test
+    public void testSetProviderOnlyOnceOnGlobal() {
+        Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers = new HashMap<>();
+        providers.put("DUMMY_PROVIDER", new DummyKeyStoreProvider());
+
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+
+        try {
+            SQLServerConnection.registerColumnEncryptionKeyStoreProviders(providers);
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+
+        assertExceptionIsThrownForGlobalCustomProviderCache(providers,
+                TestUtils.formatErrorMsg("R_CustomKeyStoreProviderSetOnce"));
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+    }
+
+    @Test
+    public void testSetProviderMoreThanOnceOnConnection() {
+        try (SQLServerConnection conn = getConnection()) {
+            conn.registerColumnEncryptionKeyStoreProvidersOnConnection(singleKeyStoreProvider);
+            assertProviderCacheContainsExpectedProviders(conn.connectionColumnEncryptionKeyStoreProvider,
+                    singleKeyStoreProvider);
+
+            conn.registerColumnEncryptionKeyStoreProvidersOnConnection(multipleKeyStoreProviders);
+            assertProviderCacheContainsExpectedProviders(conn.connectionColumnEncryptionKeyStoreProvider,
+                    multipleKeyStoreProviders);
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testSetProviderMoreThanOnceOnStatement() {
+        try (SQLServerConnection conn = getConnection();
+                SQLServerStatement stmt = (SQLServerStatement) conn.createStatement()) {
+            stmt.registerColumnEncryptionKeyStoreProvidersOnStatement(singleKeyStoreProvider);
+            assertProviderCacheContainsExpectedProviders(stmt.statementColumnEncryptionKeyStoreProviders,
+                    singleKeyStoreProvider);
+
+            stmt.registerColumnEncryptionKeyStoreProvidersOnStatement(multipleKeyStoreProviders);
+            assertProviderCacheContainsExpectedProviders(stmt.statementColumnEncryptionKeyStoreProviders,
+                    multipleKeyStoreProviders);
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    @AfterAll
+    public static void testClearUp() throws Exception {
+        SQLServerConnection.unregisterColumnEncryptionKeyStoreProviders();
+
+        if (null != globalKeyStoreProviders) {
+            SQLServerConnection.registerColumnEncryptionKeyStoreProviders(globalKeyStoreProviders);
+        }
+    }
+
+    private void assertProviderCacheContainsExpectedProviders(
+            Map<String, SQLServerColumnEncryptionKeyStoreProvider> providerCache,
+            Map<String, SQLServerColumnEncryptionKeyStoreProvider> expectedProviders) {
+        assertEquals(providerCache.size(), expectedProviders.size());
+
+        for (String key : expectedProviders.keySet()) {
+            assertTrue(providerCache.containsKey(key));
+        }
+    }
+
+    private void assertExceptionIsThrownForAllCustomProviderCaches(
+            Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers, String expectedString) {
+        assertExceptionIsThrownForGlobalCustomProviderCache(providers, expectedString);
+
+        try (SQLServerConnection conn = getConnection();
+                SQLServerStatement stmt = (SQLServerStatement) conn.createStatement()) {
+            try {
+                conn.registerColumnEncryptionKeyStoreProvidersOnConnection(providers);
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (SQLServerException ex) {
+                assertTrue(ex.getMessage().matches(expectedString));
+            }
+
+            try {
+                stmt.registerColumnEncryptionKeyStoreProvidersOnStatement(providers);
+                fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+            } catch (SQLServerException ex) {
+                assertTrue(ex.getMessage().matches(expectedString));
+            }
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+
+    }
+
+    private void assertExceptionIsThrownForGlobalCustomProviderCache(
+            Map<String, SQLServerColumnEncryptionKeyStoreProvider> providers, String expectedString) {
+        try {
+            SQLServerConnection.registerColumnEncryptionKeyStoreProviders(providers);
+            fail(TestResource.getResource("R_expectedExceptionNotThrown"));
+        } catch (SQLServerException ex) {
+            assertTrue(ex.getMessage().matches(expectedString));
+        } catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+}
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java b/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java
index c93b76f8..557e47a5 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java
@@ -196,5 +196,8 @@ public final class TestResource extends ListResourceBundle {
             {"R_socketClosed", "Socket closed"}, {"R_aeStreamReadError", "The multi-part identifier"},
             {"R_dataClassificationNotSupported", "Data Classification is not supported on this server."},
             {"R_maxResultBufferExceeded", "MaxResultBuffer exceeded {0}."},
-            {"R_databaseNotFound", "Database {0} not found."}, {"R_failedToDecrypt", "Failed to decrypt"}};
+            {"R_databaseNotFound", "Database {0} not found."}, {"R_failedToDecrypt", "Failed to decrypt"},
+            {"R_AKVProviderNull", "The Azure key store provider is null."},
+            {"R_objectNullOrEmpty", "The {0} is null or empty."},
+            {"R_cekDecryptionFailed", "Failed to decrypt a column encryption key using key store provider: {0}."}};
 }
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/TestUtils.java b/src/test/java/com/microsoft/sqlserver/jdbc/TestUtils.java
index 79903a75..697d35ea 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/TestUtils.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/TestUtils.java
@@ -477,6 +477,26 @@ public final class TestUtils {
         return sb.toString();
     }
 
+    /**
+     * 
+     * @param b
+     *        byte value
+     * @return
+     */
+    public static String byteToHexDisplayString(byte[] b) {
+        if (null == b)
+            return "(null)";
+        int hexVal;
+        StringBuilder sb = new StringBuilder(b.length * 2 + 2);
+        sb.append("0x");
+        for (byte aB : b) {
+            hexVal = aB & 0xFF;
+            sb.append(HEXCHARS[(hexVal & 0xF0) >> 4]);
+            sb.append(HEXCHARS[(hexVal & 0x0F)]);
+        }
+        return sb.toString();
+    }
+
     /**
      * conversion routine valid values 0-9 a-f A-F throws exception when failed to convert
      * 
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyColumnMappingTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyColumnMappingTest.java
index e2eeab0a..f34c5586 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyColumnMappingTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyColumnMappingTest.java
@@ -26,7 +26,6 @@ import org.junit.runner.RunWith;
 import com.microsoft.sqlserver.jdbc.ComparisonUtil;
 import com.microsoft.sqlserver.jdbc.RandomUtil;
 import com.microsoft.sqlserver.jdbc.SQLServerBulkCopy;
-import com.microsoft.sqlserver.jdbc.SQLServerBulkCopyOptions;
 import com.microsoft.sqlserver.jdbc.TestResource;
 import com.microsoft.sqlserver.jdbc.TestUtils;
 import com.microsoft.sqlserver.testframework.AbstractSQLGenerator;
@@ -44,6 +43,7 @@ import com.microsoft.sqlserver.testframework.sqlType.SqlType;
 @RunWith(JUnitPlatform.class)
 @DisplayName("BulkCopy Column Mapping Test")
 public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
+    private String prcdb = RandomUtil.getIdentifier("BulkCopy_PRC_DB");
 
     @Test
     @DisplayName("BulkCopy:test no explicit column mapping")
@@ -62,6 +62,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
                 bulkWrapper.setUsingXAConnection((0 == Constants.RANDOM.nextInt(2)) ? true : false, dsXA);
                 bulkWrapper.setUsingPooledConnection((0 == Constants.RANDOM.nextInt(2)) ? true : false, dsPool);
                 BulkCopyTestUtil.performBulkCopy(bulkWrapper, sourceTable, destTable);
+            } catch (Exception e) {
+                fail(e.getMessage());
             } finally {
                 TestUtils.dropTableIfExists(destTable.getEscapedTableName(), (Statement) stmt.product());
             }
@@ -103,6 +105,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
                     }
                 }
                 BulkCopyTestUtil.performBulkCopy(bulkWrapper, sourceTable, destTable);
+            } catch (Exception e) {
+                fail(e.getMessage());
             } finally {
                 TestUtils.dropTableIfExists(destTable.getEscapedTableName(), (Statement) stmt.product());
             }
@@ -148,6 +152,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
                     }
                 }
                 BulkCopyTestUtil.performBulkCopy(bulkWrapper, sourceTableUnicode, destTableUnicode);
+            } catch (Exception e) {
+                fail(e.getMessage());
             } finally {
                 TestUtils.dropTableIfExists(sourceTableUnicode.getEscapedTableName(), (Statement) stmt.product());
                 TestUtils.dropTableIfExists(destTableUnicode.getEscapedTableName(), (Statement) stmt.product());
@@ -213,6 +219,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
 
                     fail(form.format(msgArgs) + "\n" + destTable.getTableName() + "\n" + e.getMessage());
                 }
+            } catch (Exception e) {
+                fail(e.getMessage());
             } finally {
                 TestUtils.dropTableIfExists(sourceTable1.getEscapedTableName(), (Statement) stmt.product());
                 TestUtils.dropTableIfExists(destTable.getEscapedTableName(), (Statement) stmt.product());
@@ -254,6 +262,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
                     }
                 }
                 BulkCopyTestUtil.performBulkCopy(bulkWrapper, sourceTable, destTable, true, true);
+            } catch (Exception e) {
+                fail(e.getMessage());
             } finally {
                 TestUtils.dropTableIfExists(destTable.getEscapedTableName(), (Statement) stmt.product());
             }
@@ -361,6 +371,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
                 bulkWrapper.setUsingPooledConnection((0 == Constants.RANDOM.nextInt(2)) ? true : false, dsPool);
                 bulkWrapper.setColumnMapping(Integer.MIN_VALUE, Integer.MAX_VALUE);
                 BulkCopyTestUtil.performBulkCopy(bulkWrapper, sourceTable, destTable, true, true);
+            } catch (Exception e) {
+                fail(e.getMessage());
             } finally {
                 TestUtils.dropTableIfExists(destTable.getEscapedTableName(), (Statement) stmt.product());
             }
@@ -374,11 +386,33 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
         // Windows only as this test has problems getting dependent H2 package from Maven in Unix
         org.junit.Assume.assumeTrue(isWindows);
 
-        validateMapping("CHAR(5)", "NCHAR(5)", "фщыab");
-        validateMapping("CHAR(5)", "NVARCHAR(5)", "фщыab");
-        validateMapping("VARCHAR(5)", "NCHAR(5)", "фщыab");
-        validateMapping("VARCHAR(5)", "NVARCHAR(5)", "фщыab");
-        validateMapping("VARCHAR(5)", "NVARCHAR(max)", "фщыab");
+        validateNMapping("CHAR(5)", "NCHAR(5)", "фщыab");
+        validateNMapping("CHAR(5)", "NVARCHAR(5)", "фщыab");
+        validateNMapping("VARCHAR(5)", "NCHAR(5)", "фщыab");
+        validateNMapping("VARCHAR(5)", "NVARCHAR(5)", "фщыab");
+        validateNMapping("VARCHAR(5)", "NVARCHAR(max)", "фщыab");
+    }
+
+    @Tag(Constants.xAzureSQLDW)
+    @Tag(Constants.xAzureSQLDB)
+    @Test
+    @DisplayName("BulkCopy:test unicode char/varchar to char/varchar")
+    public void testUnicodeCharToChar() throws SQLException, ClassNotFoundException {
+
+        try (Connection con = DriverManager.getConnection(connectionString); Statement stmt = con.createStatement()) {
+            stmt.executeUpdate("CREATE DATABASE [" + prcdb + "]");
+            stmt.executeUpdate("ALTER DATABASE " + "[" + prcdb + "]" + " COLLATE Chinese_PRC_CI_AS;");
+
+            validateMapping("CHAR(10)", "CHAR(10)", "测试设计者");
+            validateMapping("CHAR(10)", "VARCHAR(10)", "测试设计者");
+            validateMapping("VARCHAR(10)", "CHAR(10)", "测试设计者");
+            validateMapping("VARCHAR(10)", "VARCHAR(10)", "测试设计者");
+            validateMapping("VARCHAR(10)", "VARCHAR(max)", "测试设计者");
+        } catch (Exception e) {
+            fail(e.getMessage());
+        } finally {
+            TestUtils.dropDatabaseIfExists(prcdb, connectionString);
+        }
     }
 
     @Tag(Constants.xAzureSQLDW)
@@ -466,7 +500,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
         }
     }
 
-    private void validateMapping(String sourceType, String destType,
+    // validate mapping from Char/Varchar to NChar/Varchar types
+    private void validateNMapping(String sourceType, String destType,
             String data) throws SQLException, ClassNotFoundException {
         Class.forName("org.h2.Driver");
         Random rand = new Random();
@@ -483,7 +518,8 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
                 sourceStmt.executeUpdate("CREATE TABLE " + sourceTable + " (col " + sourceType + ");");
                 sourceStmt.executeUpdate("INSERT INTO " + sourceTable + " VALUES('" + data + "');");
 
-                destStmt.executeUpdate("CREATE TABLE " + destTable + " (col NCHAR(5));");
+                destStmt.executeUpdate("CREATE TABLE " + destTable + " (col " + destType + ");");
+                
                 ResultSet sourceRs = sourceStmt.executeQuery("SELECT * FROM " + sourceTable);
                 bulkCopy.writeToServer(sourceRs);
 
@@ -491,6 +527,42 @@ public class BulkCopyColumnMappingTest extends BulkCopyTestSetUp {
                 destRs.next();
                 String receivedUnicodeData = destRs.getString(1);
                 assertEquals(data, receivedUnicodeData);
+            } catch (Exception e) {
+                fail(e.getMessage());
+            } finally {
+                sourceStmt.executeUpdate("DROP TABLE " + sourceTable);
+                TestUtils.dropTableIfExists(destTable, destStmt);
+            }
+        }
+    }
+
+    private void validateMapping(String sourceType, String destType,
+            String data) throws SQLException, ClassNotFoundException {
+        Random rand = new Random();
+        String sourceTable = "sourceTable" + rand.nextInt(Integer.MAX_VALUE);
+        String destTable = TestUtils
+                .escapeSingleQuotes(AbstractSQLGenerator.escapeIdentifier(RandomUtil.getIdentifier("destTable")));
+        try (Connection sourceCon = DriverManager.getConnection(connectionString + ";database=" + prcdb);
+                Connection destCon = DriverManager.getConnection(connectionString + ";database=" + prcdb);
+                Statement sourceStmt = sourceCon.createStatement(); Statement destStmt = destCon.createStatement();
+                SQLServerBulkCopy bulkCopy = new SQLServerBulkCopy(destCon)) {
+            try {
+                bulkCopy.setDestinationTableName(destTable);
+
+                sourceStmt.executeUpdate("CREATE TABLE " + sourceTable + " (col " + sourceType + ");");
+                sourceStmt.executeUpdate("INSERT INTO " + sourceTable + " VALUES('" + data + "');");
+
+                destStmt.executeUpdate("CREATE TABLE " + destTable + " (col " + destType + ");");
+
+                ResultSet sourceRs = sourceStmt.executeQuery("SELECT * FROM " + sourceTable);
+                bulkCopy.writeToServer(sourceRs);
+
+                ResultSet destRs = destStmt.executeQuery("SELECT * FROM " + destTable);
+                destRs.next();
+                String receivedUnicodeData = destRs.getString(1);
+                assertEquals(data, receivedUnicodeData);
+            } catch (Exception e) {
+                fail(e.getMessage());
             } finally {
                 sourceStmt.executeUpdate("DROP TABLE " + sourceTable);
                 TestUtils.dropTableIfExists(destTable, destStmt);
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/ISQLServerBulkRecordIssuesTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/ISQLServerBulkRecordIssuesTest.java
index cf2febbc..0752669e 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/ISQLServerBulkRecordIssuesTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/ISQLServerBulkRecordIssuesTest.java
@@ -15,6 +15,8 @@ import java.sql.SQLException;
 import java.sql.Statement;
 import java.sql.Timestamp;
 import java.text.MessageFormat;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Map;
@@ -224,6 +226,110 @@ public class ISQLServerBulkRecordIssuesTest extends AbstractTest {
         }
     }
 
+    /**
+     * Testing that sending valid values of LocalDateTime with precision 3 for datetime2 column are successful
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testSendValidValueforDatetime3ColumnAsLocalDateTime() throws Exception {
+        variation = "testSendValidValueforDatetime3ColumnAsLocalDateTime";
+        BulkData bData = new BulkData(variation);
+        query = "CREATE TABLE " + AbstractSQLGenerator.escapeIdentifier(destTable) + " (col1 datetime2(7))";
+        int counter = 0;
+        String[] result = {"2021-01-01 00:00:00.0000000", "2021-01-01 12:00:00.0000000", "2021-01-01 12:30:00.0000000",
+                "2021-01-01 12:30:44.0000000", "2021-01-01 12:30:44.0000000", "2021-01-01 12:30:44.0030000",
+                "2021-01-01 12:30:44.1000000", "2021-01-01 12:30:44.1230000", "2021-01-01 12:30:44.9990000"};
+        try (Connection con = getConnection(); Statement stmt = con.createStatement()) {
+            stmt.executeUpdate(query);
+
+            try (SQLServerBulkCopy bcOperation = new SQLServerBulkCopy(connectionString)) {
+                bcOperation.setDestinationTableName(AbstractSQLGenerator.escapeIdentifier(destTable));
+                bcOperation.writeToServer(bData);
+
+                try (ResultSet rs = stmt
+                        .executeQuery("select * from " + AbstractSQLGenerator.escapeIdentifier(destTable))) {
+                    while (rs.next()) {
+                        assertEquals(rs.getString(1), result[counter]);
+                        counter++;
+                    }
+                }
+            } catch (Exception e) {
+                fail(e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * Testing that sending valid values of LocalDateTime with precision 7 for datetime2 column are successful
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testSendValidValueforDatetime7ColumnAsLocalDateTime() throws Exception {
+        variation = "testSendValidValueforDatetime7ColumnAsLocalDateTime";
+        BulkData bData = new BulkData(variation);
+        query = "CREATE TABLE " + AbstractSQLGenerator.escapeIdentifier(destTable) + " (col1 datetime2(7))";
+        int counter = 0;
+        String[] result = {"2021-01-01 00:00:00.0000000", "2021-01-01 12:00:00.0000000", "2021-01-01 12:30:00.0000000",
+                "2021-01-01 12:30:44.0000000", "2021-01-01 12:30:44.0000000", "2021-01-01 12:30:44.0000007",
+                "2021-01-01 12:30:44.0030000", "2021-01-01 12:30:44.1000000", "2021-01-01 12:30:44.1230000",
+                "2021-01-01 12:30:44.1234567", "2021-01-01 12:30:44.9999999"};
+        try (Connection con = getConnection(); Statement stmt = con.createStatement()) {
+            stmt.executeUpdate(query);
+
+            try (SQLServerBulkCopy bcOperation = new SQLServerBulkCopy(connectionString)) {
+                bcOperation.setDestinationTableName(AbstractSQLGenerator.escapeIdentifier(destTable));
+                bcOperation.writeToServer(bData);
+
+                try (ResultSet rs = stmt
+                        .executeQuery("select * from " + AbstractSQLGenerator.escapeIdentifier(destTable))) {
+                    while (rs.next()) {
+                        assertEquals(rs.getString(1), result[counter]);
+                        counter++;
+                    }
+                }
+            } catch (Exception e) {
+                fail(e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * Testing that sending valid values of LocalTime for datetime2 column are successful
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testSendValidValueforDatetime2ColumnAsLocalTime() throws Exception {
+        variation = "testSendValidValueforDatetime2ColumnAsLocalTime";
+        BulkData bData = new BulkData(variation);
+        query = "CREATE TABLE " + AbstractSQLGenerator.escapeIdentifier(destTable) + " (col1 datetime2(7))";
+        int counter = 0;
+        String[] result = {"1900-01-01 00:00:00.0000000", "1900-01-01 12:00:00.0000000", "1900-01-01 12:30:00.0000000",
+                "1900-01-01 12:30:44.0000000", "1900-01-01 12:30:44.0000007", "1900-01-01 12:30:44.0030000",
+                "1900-01-01 12:30:44.1000000", "1900-01-01 12:30:44.1230000", "1900-01-01 12:30:44.1234567",
+                "1900-01-01 12:30:44.9999999"};
+        try (Connection con = getConnection(); Statement stmt = con.createStatement()) {
+            stmt.executeUpdate(query);
+
+            try (SQLServerBulkCopy bcOperation = new SQLServerBulkCopy(connectionString)) {
+                bcOperation.setDestinationTableName(AbstractSQLGenerator.escapeIdentifier(destTable));
+                bcOperation.writeToServer(bData);
+
+                try (ResultSet rs = stmt
+                        .executeQuery("select * from " + AbstractSQLGenerator.escapeIdentifier(destTable))) {
+                    while (rs.next()) {
+                        assertEquals(rs.getString(1), result[counter]);
+                        counter++;
+                    }
+                }
+            } catch (Exception e) {
+                fail(e.getMessage());
+            }
+        }
+    }
+
     /**
      * Prepare test
      * 
@@ -273,6 +379,9 @@ class BulkData implements ISQLServerBulkData {
 
     Map<Integer, ColumnMetadata> columnMetadata;
     ArrayList<Timestamp> dateData;
+    ArrayList<LocalDateTime> datetime3Data;
+    ArrayList<LocalDateTime> datetime7Data;
+    ArrayList<LocalTime> timeData;
     ArrayList<String> stringData;
     ArrayList<byte[]> byteData;
 
@@ -307,7 +416,6 @@ class BulkData implements ISQLServerBulkData {
             dateData = new ArrayList<>();
             dateData.add(Timestamp.valueOf("1954-05-22 02:43:37.1234"));
             rowCount = dateData.size();
-
         } else if (variation.equalsIgnoreCase("testBinaryColumnAsByte")) {
             isStringData = false;
             columnMetadata = new HashMap<>();
@@ -317,7 +425,6 @@ class BulkData implements ISQLServerBulkData {
             byteData = new ArrayList<>();
             byteData.add("helloo".getBytes());
             rowCount = byteData.size();
-
         } else if (variation.equalsIgnoreCase("testBinaryColumnAsString")) {
             isStringData = true;
             columnMetadata = new HashMap<>();
@@ -327,10 +434,7 @@ class BulkData implements ISQLServerBulkData {
             stringData = new ArrayList<>();
             stringData.add("616368697412");
             rowCount = stringData.size();
-
-        }
-
-        else if (variation.equalsIgnoreCase("testSendValidValueforBinaryColumnAsString")) {
+        } else if (variation.equalsIgnoreCase("testSendValidValueforBinaryColumnAsString")) {
             isStringData = true;
             columnMetadata = new HashMap<>();
 
@@ -339,10 +443,63 @@ class BulkData implements ISQLServerBulkData {
             stringData = new ArrayList<>();
             stringData.add("010101");
             rowCount = stringData.size();
+        } else if (variation.equalsIgnoreCase("testSendValidValueforDatetime3ColumnAsLocalDateTime")) {
+            isStringData = false;
+            columnMetadata = new HashMap<>();
+
+            columnMetadata.put(1, new ColumnMetadata("datetime2(3)", java.sql.Types.TIMESTAMP, 23, 0));
+
+            datetime3Data = new ArrayList<>();
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 0, 0));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 0));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 30));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 0));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 3000000));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 100000000));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 123000000));
+            datetime3Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 999000000));
+            rowCount = datetime3Data.size();
+        } else if (variation.equalsIgnoreCase("testSendValidValueforDatetime7ColumnAsLocalDateTime")) {
+            isStringData = false;
+            columnMetadata = new HashMap<>();
 
+            columnMetadata.put(1, new ColumnMetadata("datetime2(7)", java.sql.Types.TIMESTAMP, 27, 0));
+
+            datetime7Data = new ArrayList<>();
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 0, 0));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 0));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 0));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 700));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 3000000));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 100000000));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 123000000));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 123456700));
+            datetime7Data.add(LocalDateTime.of(2021, 1, 1, 12, 30, 44, 999999900));
+            rowCount = datetime7Data.size();
+        } else if (variation.equalsIgnoreCase("testSendValidValueforDatetime2ColumnAsLocalTime")) {
+            isStringData = false;
+            columnMetadata = new HashMap<>();
+
+            columnMetadata.put(1, new ColumnMetadata("datetime2(7)", java.sql.Types.TIME, 16, 0));
+
+            timeData = new ArrayList<>();
+            timeData.add(LocalTime.of(0, 0));
+            timeData.add(LocalTime.of(12, 0));
+            timeData.add(LocalTime.of(12, 30));
+            timeData.add(LocalTime.of(12, 30, 44));
+            timeData.add(LocalTime.of(12, 30, 44, 700));
+            timeData.add(LocalTime.of(12, 30, 44, 3000000));
+            timeData.add(LocalTime.of(12, 30, 44, 100000000));
+            timeData.add(LocalTime.of(12, 30, 44, 123000000));
+            timeData.add(LocalTime.of(12, 30, 44, 123456700));
+            timeData.add(LocalTime.of(12, 30, 44, 999999900));
+            rowCount = timeData.size();
         }
-        counter = 0;
 
+        counter = 0;
     }
 
     @Override
@@ -378,6 +535,12 @@ class BulkData implements ISQLServerBulkData {
         else {
             if (null != dateData)
                 dataRow[0] = dateData.get(counter);
+            else if (null != datetime3Data)
+                dataRow[0] = datetime3Data.get(counter);
+            else if (null != datetime7Data)
+                dataRow[0] = datetime7Data.get(counter);
+            else if (null != timeData)
+                dataRow[0] = timeData.get(counter);
             else if (null != byteData)
                 dataRow[0] = byteData.get(counter);
         }
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/connection/RequestBoundaryMethodsTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/connection/RequestBoundaryMethodsTest.java
index 7b34707b..c3507fc2 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/connection/RequestBoundaryMethodsTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/connection/RequestBoundaryMethodsTest.java
@@ -488,6 +488,7 @@ public class RequestBoundaryMethodsTest extends AbstractTest {
         verifiedMethodNames.add("createArrayOf");
         verifiedMethodNames.add("setUseFmtOnly");
         verifiedMethodNames.add("setDelayLoadingLobs");
+        verifiedMethodNames.add("registerColumnEncryptionKeyStoreProvidersOnConnection");
 
         return verifiedMethodNames;
     }
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/unit/statement/StatementTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/unit/statement/StatementTest.java
index 338a688f..f17fb07a 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/unit/statement/StatementTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/unit/statement/StatementTest.java
@@ -7,6 +7,7 @@ package com.microsoft.sqlserver.jdbc.unit.statement;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.fail;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.StringReader;
diff --git a/src/test/java/com/microsoft/sqlserver/testframework/Constants.java b/src/test/java/com/microsoft/sqlserver/testframework/Constants.java
index 57d7e784..e7060e5b 100644
--- a/src/test/java/com/microsoft/sqlserver/testframework/Constants.java
+++ b/src/test/java/com/microsoft/sqlserver/testframework/Constants.java
@@ -98,6 +98,7 @@ public final class Constants {
     public static final String AZURE_KEY_VAULT_NAME = "AZURE_KEY_VAULT";
     public static final String JAVA_KEY_STORE_NAME = "MSSQL_JAVA_KEYSTORE";
     public static final String CUSTOM_KEYSTORE_NAME = "CUSTOM_KEYSTORE";
+    public static final String DUMMY_KEYSTORE_NAME = "DUMMY_PROVIDER";
     public static final String JAVA_KEY_STORE_FILENAME = "JavaKeyStore.txt";
     public static final String JAVA_KEY_STORE_SECRET = "JavaKeyStorePassword";
     public static final String JKS = "JKS";
diff --git a/src/test/java/com/microsoft/sqlserver/testframework/DummyKeyStoreProvider.java b/src/test/java/com/microsoft/sqlserver/testframework/DummyKeyStoreProvider.java
new file mode 100644
index 00000000..fbfa3fda
--- /dev/null
+++ b/src/test/java/com/microsoft/sqlserver/testframework/DummyKeyStoreProvider.java
@@ -0,0 +1,48 @@
+/*
+ * Microsoft JDBC Driver for SQL Server Copyright(c) 2016 Microsoft Corporation All rights reserved. This program is
+ * made available under the terms of the MIT License. See the LICENSE file in the project root for more information.
+ */
+package com.microsoft.sqlserver.testframework;
+
+import com.microsoft.sqlserver.jdbc.SQLServerColumnEncryptionKeyStoreProvider;
+import com.microsoft.sqlserver.jdbc.SQLServerException;
+
+
+/**
+ * DummyKeyStoreProvider Class for Tests. 
+ * 
+ */
+public class DummyKeyStoreProvider extends SQLServerColumnEncryptionKeyStoreProvider {
+    public String name = "DUMMY_PROVIDER";
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return this.name;
+    }
+
+    @Override
+    public byte[] decryptColumnEncryptionKey(String masterKeyPath, String encryptionAlgorithm,
+            byte[] encryptedColumnEncryptionKey) throws SQLServerException {
+        // Not implemented
+        throw new UnsupportedOperationException();
+            }
+
+    @Override
+    public byte[] encryptColumnEncryptionKey(String masterKeyPath,
+            String encryptionAlgorithm,
+            byte[] columnEncryptionKey) throws SQLServerException {
+        // Not implemented
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public boolean verifyColumnMasterKeyMetadata(String masterKeyPath,
+            boolean allowEnclaveComputations,
+            byte[] signature) throws SQLServerException {
+        // Not implemented
+        throw new UnsupportedOperationException();
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/com/microsoft/sqlserver/testframework/sqlType/SqlFloat.java b/src/test/java/com/microsoft/sqlserver/testframework/sqlType/SqlFloat.java
index cdf227df..656cd48e 100644
--- a/src/test/java/com/microsoft/sqlserver/testframework/sqlType/SqlFloat.java
+++ b/src/test/java/com/microsoft/sqlserver/testframework/sqlType/SqlFloat.java
@@ -26,10 +26,13 @@ public class SqlFloat extends SqlType {
 
     public Object createdata() {
         // for float in SQL Server, any precision <=24 is considered as real so the value must be within
-        // SqlTypeValue.REAL.minValue/maxValue
+        // SqlTypeValue.REAL.minValue/maxValue however this needs to be bounded for nextDouble origin and bound params
         if (precision > 24) {
+            minvalue = ((Double) minvalue < Double.MIN_VALUE) ? Double.MIN_VALUE : minvalue;
+            maxvalue = ((Double) maxvalue > Double.MAX_VALUE) ? Double.MAX_VALUE : maxvalue;
             return ThreadLocalRandom.current().nextDouble(((Double) minvalue), ((Double) maxvalue));
         } else {
+
             return ThreadLocalRandom.current().nextDouble((Float) SqlTypeValue.REAL.minValue,
                     (Float) SqlTypeValue.REAL.maxValue);
         }