Repository: microsoft/mssql-jdbc
Pull request #2338 (2024-02-22 00:58:45)

Finish support for RFC4180 for CSV bulk insert operations
RFC4180 specifies support for fields that contain newlines as long as they are quoted.  See section 6 in the specification here: https://www.ietf.org/rfc/rfc4180.txt

This PR enhances the read operation when the option escapeColumnDelimitersCSV is set to true to allow newline characters to survive the CSV parse process and find their way into the record.  An existing unit test was enhanced to test the behavior.


microsoft-github-policy-service[bot]: 
@funkyjive please read the following Contributor License Agreement(CLA). If you agree with the CLA, please reply with the following information.

```
@microsoft-github-policy-service agree [company="{your company}"]
```

> Options:
> - (default - no company specified) I have sole ownership of intellectual property rights to my Submissions and I am not making Submissions in the course of work for my employer.
>```
>@microsoft-github-policy-service agree
>```
> - (when company given) I am making Submissions in the course of work for my employer (or my employer has intellectual property rights in my Submissions by contract or applicable law). I have permission from my employer to make Submissions and enter into this Agreement on behalf of my employer. By signing below, the defined term “You” includes me and my employer.
>```
>@microsoft-github-policy-service agree company="Microsoft"
>```

<details>
    <summary>Contributor License Agreement</summary>
    
### Contribution License Agreement
This Contribution License Agreement (**“Agreement”**) is agreed to by the party signing below (**“You”**),
and conveys certain license rights to Microsoft Corporation and its affiliates (“Microsoft”) for Your
contributions to Microsoft open source projects. This Agreement is effective as of the latest signature
date below.
1. **Definitions**.
**“Code”** means the computer software code, whether in human-readable or machine-executable form,
that is delivered by You to Microsoft under this Agreement.
**“Project”** means any of the projects owned or managed by Microsoft and offered under a license
approved by the Open Source Initiative (www.opensource.org).
**“Submit”** is the act of uploading, submitting, transmitting, or distributing code or other content to any
Project, including but not limited to communication on electronic mailing lists, source code control
systems, and issue tracking systems that are managed by, or on behalf of, the Project for the purpose of
discussing and improving that Project, but excluding communication that is conspicuously marked or
otherwise designated in writing by You as “Not a Submission.”
**“Submission”** means the Code and any other copyrightable material Submitted by You, including any
associated comments and documentation.
2. **Your Submission**. You must agree to the terms of this Agreement before making a Submission to any
Project. This Agreement covers any and all Submissions that You, now or in the future (except as
described in Section 4 below), Submit to any Project.
3. **Originality of Work**. You represent that each of Your Submissions is entirely Your original work.
Should You wish to Submit materials that are not Your original work, You may Submit them separately
to the Project if You (a) retain all copyright and license information that was in the materials as You
received them, (b) in the description accompanying Your Submission, include the phrase “Submission
containing materials of a third party:” followed by the names of the third party and any licenses or other
restrictions of which You are aware, and (c) follow any other instructions in the Project’s written
guidelines concerning Submissions.
4. **Your Employer**. References to “employer” in this Agreement include Your employer or anyone else
for whom You are acting in making Your Submission, e.g. as a contractor, vendor, or agent. If Your
Submission is made in the course of Your work for an employer or Your employer has intellectual
property rights in Your Submission by contract or applicable law, You must secure permission from Your
employer to make the Submission before signing this Agreement. In that case, the term “You” in this
Agreement will refer to You and the employer collectively. If You change employers in the future and
desire to Submit additional Submissions for the new employer, then You agree to sign a new Agreement
and secure permission from the new employer before Submitting those Submissions.
5. **Licenses**.
  * **Copyright License**. You grant Microsoft, and those who receive the Submission directly or
indirectly from Microsoft, a perpetual, worldwide, non-exclusive, royalty-free, irrevocable license in the
Submission to reproduce, prepare derivative works of, publicly display, publicly perform, and distribute
the Submission and such derivative works, and to sublicense any or all of the foregoing rights to third
parties.
  * **Patent License**. You grant Microsoft, and those who receive the Submission directly or
indirectly from Microsoft, a perpetual, worldwide, non-exclusive, royalty-free, irrevocable license under
Your patent claims that are necessarily infringed by the Submission or the combination of the
Submission with the Project to which it was Submitted to make, have made, use, offer to sell, sell and
import or otherwise dispose of the Submission alone or with the Project.
  * **Other Rights Reserved**. Each party reserves all rights not expressly granted in this Agreement.
No additional licenses or rights whatsoever (including, without limitation, any implied licenses) are
granted by implication, exhaustion, estoppel or otherwise.
6. **Representations and Warranties**. You represent that You are legally entitled to grant the above
licenses. You represent that each of Your Submissions is entirely Your original work (except as You may
have disclosed under Section 3). You represent that You have secured permission from Your employer to
make the Submission in cases where Your Submission is made in the course of Your work for Your
employer or Your employer has intellectual property rights in Your Submission by contract or applicable
law. If You are signing this Agreement on behalf of Your employer, You represent and warrant that You
have the necessary authority to bind the listed employer to the obligations contained in this Agreement.
You are not expected to provide support for Your Submission, unless You choose to do so. UNLESS
REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING, AND EXCEPT FOR THE WARRANTIES
EXPRESSLY STATED IN SECTIONS 3, 4, AND 6, THE SUBMISSION PROVIDED UNDER THIS AGREEMENT IS
PROVIDED WITHOUT WARRANTY OF ANY KIND, INCLUDING, BUT NOT LIMITED TO, ANY WARRANTY OF
NONINFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
7. **Notice to Microsoft**. You agree to notify Microsoft in writing of any facts or circumstances of which
You later become aware that would make Your representations in this Agreement inaccurate in any
respect.
8. **Information about Submissions**. You agree that contributions to Projects and information about
contributions may be maintained indefinitely and disclosed publicly, including Your name and other
information that You submit with Your Submission.
9. **Governing Law/Jurisdiction**. This Agreement is governed by the laws of the State of Washington, and
the parties consent to exclusive jurisdiction and venue in the federal courts sitting in King County,
Washington, unless no federal subject matter jurisdiction exists, in which case the parties consent to
exclusive jurisdiction and venue in the Superior Court of King County, Washington. The parties waive all
defenses of lack of personal jurisdiction and forum non-conveniens.
10. **Entire Agreement/Assignment**. This Agreement is the entire agreement between the parties, and
supersedes any and all prior agreements, understandings or communications, written or oral, between
the parties relating to the subject matter hereof. This Agreement may be assigned by Microsoft.

</details>
<!-- Policy app identification https://img.shields.io/static/v1?label=ContributorLicenseAgreement. -->

Patch:
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/ActivityCorrelator.java b/src/main/java/com/microsoft/sqlserver/jdbc/ActivityCorrelator.java
index 974ac3c4..072d34f5 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/ActivityCorrelator.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/ActivityCorrelator.java
@@ -5,26 +5,32 @@
 
 package com.microsoft.sqlserver.jdbc;
 
+import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 
 
 /**
- * ActivityCorrelator provides the APIs to access the ActivityId in TLS
+ * ActivityCorrelator provides the APIs to access the ActivityId in a map
  */
 final class ActivityCorrelator {
 
-    private static ThreadLocal<ActivityId> t_ActivityId = new ThreadLocal<ActivityId>() {
-        @Override
-        protected ActivityId initialValue() {
-            return new ActivityId();
-        }
-    };
+    private static Map<Long, ActivityId> activityIdMap = new ConcurrentHashMap<Long, ActivityId>();
 
     static ActivityId getCurrent() {
-        return t_ActivityId.get();
+        // get the value, not reference
+        @SuppressWarnings("deprecation")
+        long uniqueThreadId = Thread.currentThread().getId();
+
+        // Since the Id for each thread is unique, this assures that the below if statement is run only once per thread.
+        if (!activityIdMap.containsKey(uniqueThreadId)) {
+            activityIdMap.put(uniqueThreadId, new ActivityId());
+        }
+
+        return activityIdMap.get(uniqueThreadId);
     }
 
-    // Increment the Sequence number of the ActivityId in TLS
+    // Increment the Sequence number of the ActivityId
     // and return the ActivityId with new Sequence number
     static ActivityId getNext() {
         return getCurrent().getIncrement();
@@ -34,6 +40,16 @@ final class ActivityCorrelator {
      * Prevent instantiation.
      */
     private ActivityCorrelator() {}
+
+    static void cleanupActivityId() {
+        // remove the ActivityId that belongs to this thread.
+        @SuppressWarnings("deprecation")
+        long uniqueThreadId = Thread.currentThread().getId();
+
+        if (activityIdMap.containsKey(uniqueThreadId)) {
+            activityIdMap.remove(uniqueThreadId);
+        }
+    }
 }
 
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java b/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java
index e5e175b9..afde3910 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/IOBuffer.java
@@ -1440,7 +1440,7 @@ final class TDSChannel implements Serializable {
         }
 
         @Override
-        public synchronized int getReceiveBufferSize() throws SocketException {
+        public int getReceiveBufferSize() throws SocketException {
             return tdsChannel.tcpSocket.getReceiveBufferSize();
         }
 
@@ -1455,7 +1455,7 @@ final class TDSChannel implements Serializable {
         }
 
         @Override
-        public synchronized int getSendBufferSize() throws SocketException {
+        public int getSendBufferSize() throws SocketException {
             return tdsChannel.tcpSocket.getSendBufferSize();
         }
 
@@ -1465,7 +1465,7 @@ final class TDSChannel implements Serializable {
         }
 
         @Override
-        public synchronized int getSoTimeout() throws SocketException {
+        public int getSoTimeout() throws SocketException {
             return tdsChannel.tcpSocket.getSoTimeout();
         }
 
@@ -1536,19 +1536,19 @@ final class TDSChannel implements Serializable {
         // Ignore calls to methods that would otherwise allow the SSL socket
         // to directly manipulate the underlying TCP socket
         @Override
-        public synchronized void close() throws IOException {
+        public void close() throws IOException {
             if (logger.isLoggable(Level.FINER))
                 logger.finer(logContext + " Ignoring close");
         }
 
         @Override
-        public synchronized void setReceiveBufferSize(int size) throws SocketException {
+        public void setReceiveBufferSize(int size) throws SocketException {
             if (logger.isLoggable(Level.FINER))
                 logger.finer(toString() + " Ignoring setReceiveBufferSize size:" + size);
         }
 
         @Override
-        public synchronized void setSendBufferSize(int size) throws SocketException {
+        public void setSendBufferSize(int size) throws SocketException {
             if (logger.isLoggable(Level.FINER))
                 logger.finer(toString() + " Ignoring setSendBufferSize size:" + size);
         }
@@ -1566,7 +1566,7 @@ final class TDSChannel implements Serializable {
         }
 
         @Override
-        public synchronized void setSoTimeout(int timeout) throws SocketException {
+        public void setSoTimeout(int timeout) throws SocketException {
             tdsChannel.tcpSocket.setSoTimeout(timeout);
         }
 
@@ -2366,6 +2366,10 @@ final class TDSChannel implements Serializable {
     final void setNetworkTimeout(int timeout) throws IOException {
         tcpSocket.setSoTimeout(timeout);
     }
+
+    void resetTcpSocketTimeout() throws SocketException {
+        this.tcpSocket.setSoTimeout(con.getSocketTimeoutMilliseconds());
+    }
 }
 
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerConnection.java b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerConnection.java
index 2f54a98a..2479e506 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerConnection.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerConnection.java
@@ -483,6 +483,25 @@ public interface ISQLServerConnection extends java.sql.Connection {
      */
     void setAccessTokenCallbackClass(String accessTokenCallbackClass);
 
+    /**
+     * Get Currently installed message handler on the connection
+     * 
+     * @see ISQLServerMessageHandler#messageHandler(ISQLServerMessage)
+     * @return ISQLServerMessageHandler
+     */
+    ISQLServerMessageHandler getServerMessageHandler();
+
+    /**
+     * Set message handler on the connection
+     * 
+     * @param messageHandler
+     *        message handler
+     * 
+     * @see ISQLServerMessageHandler#messageHandler(ISQLServerMessage)
+     * @return ISQLServerMessageHandler
+     */
+    ISQLServerMessageHandler setServerMessageHandler(ISQLServerMessageHandler messageHandler);
+
     /**
      * Returns the current flag for calcBigDecimalPrecision.
      *
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerMessage.java b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerMessage.java
new file mode 100644
index 00000000..622b865c
--- /dev/null
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerMessage.java
@@ -0,0 +1,99 @@
+/*
+ * 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.sql.SQLException;
+
+
+/**
+ * Provides an interface SQLServerMessage
+ */
+public interface ISQLServerMessage {
+    /**
+     * Returns SQLServerError containing detailed info about SQL Server Message as received from SQL Server.
+     * 
+     * @return SQLServerError
+     */
+    public SQLServerError getSQLServerMessage();
+
+    /**
+     * Returns error message as received from SQL Server
+     * 
+     * @return Error Message
+     */
+    public String getErrorMessage();
+
+    /**
+     * Returns error number as received from SQL Server
+     * 
+     * @return Error Number
+     */
+    public int getErrorNumber();
+
+    /**
+     * Returns error state as received from SQL Server
+     * 
+     * @return Error State
+     */
+    public int getErrorState();
+
+    /**
+     * Returns Severity of error (as int value) as received from SQL Server
+     * 
+     * @return Error Severity
+     */
+    public int getErrorSeverity();
+
+    /**
+     * Returns name of the server where exception occurs as received from SQL Server
+     * 
+     * @return Server Name
+     */
+    public String getServerName();
+
+    /**
+     * Returns name of the stored procedure where exception occurs as received from SQL Server
+     * 
+     * @return Procedure Name
+     */
+    public String getProcedureName();
+
+    /**
+     * Returns line number where the error occurred in Stored Procedure returned by <code>getProcedureName()</code> as
+     * received from SQL Server
+     * 
+     * @return Line Number
+     */
+    public long getLineNumber();
+
+    /**
+     * Creates a SQLServerException or SQLServerWarning from this SQLServerMessage<br>
+     * 
+     * @return
+     *         <ul>
+     *         <li>SQLServerException if it's a SQLServerError object</li>
+     *         <li>SQLServerWarning if it's a SQLServerInfoMessage object</li>
+     *         </ul>
+     */
+    public SQLException toSqlExceptionOrSqlWarning();
+
+    /**
+     * Check if this is a isErrorMessage
+     * 
+     * @return true if it's an instance of SQLServerError
+     */
+    public default boolean isErrorMessage() {
+        return this instanceof SQLServerError;
+    }
+
+    /**
+     * Check if this is a SQLServerInfoMessage
+     * 
+     * @return true if it's an instance of SQLServerInfoMessage
+     */
+    public default boolean isInfoMessage() {
+        return this instanceof SQLServerInfoMessage;
+    }
+}
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerMessageHandler.java b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerMessageHandler.java
new file mode 100644
index 00000000..2f0b71de
--- /dev/null
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/ISQLServerMessageHandler.java
@@ -0,0 +1,103 @@
+/*
+ * 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;
+
+/**
+ * You can use the ISQLServerMessageHandler interface to customize the way JDBC handles error messages generated by the SQL Server.
+ * Implementing ISQLServerMessageHandler in your own class for handling error messages can provide the following benefits:
+ * <ul>
+ * <li><b>"message feedback"</b><br>
+ * Display Server messages from a long running SQL Statement<br>
+ * Like <code>RAISERROR ('Progress message...', 0, 1) WITH NOWAIT</code><br>
+ * Or Status messages from a running backup...<br>
+ * </li>
+ * <li><b>"Universal" error logging</b><br>
+ * Your error-message handler can contain the logic for handling all error logging.
+ * </li>
+ * <li><b>"Universal" error handling</b><br>
+ * Error-handling logic can be placed in your error-message handler, instead of being repeated throughout your application.
+ * </li>
+ * <li><b>Remapping of error-message severity</b>, based on application requirements<br>
+ * Your error-message handler can contain logic for recognizing specific error messages, and downgrading or upgrading
+ * their severity based on application considerations rather than the severity rating of the server.
+ * For example, during a cleanup operation that deletes old rows, you might want to downgrade the severity of a
+ * message that a row does not exist. However, you may want to upgrade the severity in other circumstances.
+ * </li>
+ * </ul>
+ * <p>
+ * For example code, see {@link #messageHandler(ISQLServerMessage)}
+ */
+public interface ISQLServerMessageHandler {
+    /**
+     * You can use the ISQLServerMessageHandler interface to customize the way JDBC handles error messages generated by the SQL Server.
+     * Implementing ISQLServerMessageHandler in your own class for handling error messages can provide the following benefits:
+     * <ul>
+     * <li><b>"message feedback"</b><br>
+     * Display Server messages from a long running SQL Statement<br>
+     * Like <code>RAISERROR ('Progress message...', 0, 1) WITH NOWAIT</code><br>
+     * Or Status messages from a running backup...<br>
+     * </li>
+     * <li><b>"Universal" error logging</b><br>
+     * Your error-message handler can contain the logic for handling all error logging.
+     * </li>
+     * <li><b>"Universal" error handling</b><br>
+     * Error-handling logic can be placed in your error-message handler, instead of being repeated throughout your application.
+     * </li>
+     * <li><b>Remapping of error-message severity</b>, based on application requirements<br>
+     * Your error-message handler can contain logic for recognizing specific error messages, and downgrading or upgrading
+     * their severity based on application considerations rather than the severity rating of the server.
+     * For example, during a cleanup operation that deletes old rows, you might want to downgrade the severity of a
+     * message that a row does not exist. However, you may want to upgrade the severity in other circumstances.
+     * </li>
+     * </ul>
+     * 
+     * Example code:
+     * 
+     * <pre>
+     * public ISQLServerMessage messageHandler(ISQLServerMessage serverErrorOrWarning) {
+     *     ISQLServerMessage retObj = serverErrorOrWarning;
+     *
+     *     if (serverErrorOrWarning.isErrorMessage()) {
+     *
+     *         // Downgrade: 2601 -- Cannot insert duplicate key row...
+     *         if (2601 == serverErrorOrWarning.getErrorNumber()) {
+     *             retObj = serverErrorOrWarning.getSQLServerMessage().toSQLServerInfoMessage();
+     *         }
+     *
+     *         // Discard: 3701 -- Cannot drop the table ...
+     *         if (3701 == serverErrorOrWarning.getErrorNumber()) {
+     *             retObj = null;
+     *         }
+     *     }
+     *
+     *     return retObj;
+     * }
+     * 
+     * </pre>
+     * 
+     * @param serverErrorOrWarning
+     *        server error or warning
+     * @return
+     *         <ul>
+     *         <li><b>unchanged</b> same object as passed in.<br>
+     *         The JDBC driver will work as if no message hander was installed<br>
+     *         Possibly used for logging functionality<br>
+     *         </li>
+     *         <li><b>null</b><br>
+     *         The JDBC driver will <i>discard</i> this message. No SQLException will be thrown
+     *         </li>
+     *         <li><b>SQLServerInfoMessage</b> object<br>
+     *         Create a "SQL warning" from a input database error, and return it.
+     *         This results in the warning being added to the warning-message chain.
+     *         </li>
+     *         <li><b>SQLServerError</b> object<br>
+     *         If the originating message is a SQL warning (SQLServerInfoMessage object), messageHandler can evaluate
+     *         the SQL warning as urgent and create and return a SQL exception (SQLServerError object)
+     *         to be thrown once control is returned to the JDBC Driver.
+     *         </li>
+     *         </ul>
+     */
+    ISQLServerMessage messageHandler(ISQLServerMessage serverErrorOrWarning);
+}
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/PersistentTokenCacheAccessAspect.java b/src/main/java/com/microsoft/sqlserver/jdbc/PersistentTokenCacheAccessAspect.java
index 31b87fb2..24458a7f 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/PersistentTokenCacheAccessAspect.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/PersistentTokenCacheAccessAspect.java
@@ -22,13 +22,16 @@ import java.util.concurrent.locks.ReentrantLock;
  * @see <a href="https://aka.ms/msal4j-token-cache">https://aka.ms/msal4j-token-cache</a>
  */
 public class PersistentTokenCacheAccessAspect implements ITokenCacheAccessAspect {
-    private static PersistentTokenCacheAccessAspect instance = new PersistentTokenCacheAccessAspect();
-
+    private static PersistentTokenCacheAccessAspect instance;
     private final Lock lock = new ReentrantLock();
 
-    private PersistentTokenCacheAccessAspect() {}
+    static final long TIME_TO_LIVE = 86400000L; // Token cache time to live (24 hrs).
+    private long expiryTime;
 
     static PersistentTokenCacheAccessAspect getInstance() {
+        if (instance == null) {
+            instance = new PersistentTokenCacheAccessAspect();
+        }
         return instance;
     }
 
@@ -62,6 +65,25 @@ public class PersistentTokenCacheAccessAspect implements ITokenCacheAccessAspect
 
     }
 
+    /**
+     * Get expiry time
+     * 
+     * @return expiry time
+     */
+    public long getExpiryTime() {
+        return this.expiryTime;
+    }
+
+    /**
+     * Set expiry time
+     * 
+     * @param expiryTime
+     *        expiry time
+     */
+    public void setExpiryTime(long expiryTime) {
+        this.expiryTime = expiryTime;
+    }
+
     /**
      * Clears User token cache. This will clear all account info so interactive login will be required on the next
      * request to acquire an access token.
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCSVFileRecord.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCSVFileRecord.java
index 195b197e..25f99214 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCSVFileRecord.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerBulkCSVFileRecord.java
@@ -212,6 +212,47 @@ public class SQLServerBulkCSVFileRecord extends SQLServerBulkRecord implements j
         }
     }
 
+    /*
+     * RFC4180 specifies that rules for quoted fields. It allows quoted string data to contain newlines data
+     * provided the contents otherwise conforms to the rules for escaping quotes. For example, the following is valid:
+     * "a","b","c"
+     * "aaa","b <-- newline is retained in data field
+     * bb","c"
+     * "aa","bb","cc"
+     * We cannot simply use fileReader.readLine() to read these records but instead must continue reading until we reach
+     * a newline that is not contained within quotes.
+     */
+    private String readLineEscapeDelimiters() throws SQLServerException {
+        int quoteCount = 0;
+        StringBuilder sb = new StringBuilder();
+        try {
+            int c;
+            while ((c = fileReader.read()) != -1) {
+                if ((c == '\n' || c == '\r') && quoteCount % 2 == 0) { // newlines only end the record if we are not in quotes
+                    fileReader.mark(1);
+                    c = fileReader.read(); // we might have read \r of a \r\n, if so we need to read the \n as well
+                    if (c != '\n') {
+                        fileReader.reset(); // only delimited by \n, unread last char so it goes into the next record
+                    }
+                    break;
+                }
+                sb.append((char) c);
+                if (c == '"') {
+                    quoteCount++;
+                }
+            }
+            if (c == -1 && quoteCount % 2 != 0) { // stream ended, but we are within quotes -- data problem
+                throw new SQLServerException(SQLServerException.getErrString("R_InvalidCSVQuotes"), null, 0, null);
+            }
+            if (c == -1) { // keep semantics of readLine() by returning a null when there is no more data
+                return null;
+            }
+        } catch (IOException e) {
+            throw new SQLServerException(e.getMessage(), null, 0, e);
+        }
+        return sb.toString();
+    }
+
     private void initLoggerResources() {
         super.loggerPackageName = "com.microsoft.sqlserver.jdbc.SQLServerBulkCSVFileRecord";
     }
@@ -526,7 +567,7 @@ public class SQLServerBulkCSVFileRecord extends SQLServerBulkRecord implements j
     @Override
     public boolean next() throws SQLServerException {
         try {
-            currentLine = fileReader.readLine();
+            currentLine = escapeDelimiters ? readLineEscapeDelimiters() : fileReader.readLine();
         } catch (IOException e) {
             throw new SQLServerException(e.getMessage(), null, 0, e);
         }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java
index 89a433f0..42281688 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionAzureKeyVaultProvider.java
@@ -919,6 +919,22 @@ public class SQLServerColumnEncryptionAzureKeyVaultProvider extends SQLServerCol
         return isValid;
     }
 
+    /**
+     * Sign column master key metadata
+     * 
+     * @param masterKeyPath
+     *        master key path
+     * 
+     * @param allowEnclaveComputations
+     *        flag whether to allow enclave computations
+     * 
+     * @return
+     *         column master key metadata
+     * 
+     * @throws SQLServerException
+     *         when an error occurs
+     * 
+     */
     public byte[] signColumnMasterKeyMetadata(String masterKeyPath,
             boolean allowEnclaveComputations) throws SQLServerException {
         if (!allowEnclaveComputations) {
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionJavaKeyStoreProvider.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionJavaKeyStoreProvider.java
index 50359f04..d2e2f28c 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionJavaKeyStoreProvider.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerColumnEncryptionJavaKeyStoreProvider.java
@@ -164,6 +164,22 @@ public class SQLServerColumnEncryptionJavaKeyStoreProvider extends SQLServerColu
         return isValid;
     }
 
+    /**
+     * Sign column master key metadata
+     * 
+     * @param masterKeyPath
+     *        master key path
+     * 
+     * @param allowEnclaveComputations
+     *        flag whether to allow enclave computations
+     * 
+     * @return
+     *         column master key metadata
+     * 
+     * @throws SQLServerException
+     *         when an error occurs
+     * 
+     */
     public byte[] signColumnMasterKeyMetadata(String masterKeyPath,
             boolean allowEnclaveComputations) throws SQLServerException {
         if (!allowEnclaveComputations)
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java
index afe161d3..12796539 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnection.java
@@ -1053,6 +1053,9 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
         this.ignoreOffsetOnDateTimeOffsetConversion = ignoreOffsetOnDateTimeOffsetConversion;
     }
 
+    /**
+     * Flag to indicate whether the driver should calculate precision for BigDecimal inputs, as opposed to using the maximum allowed valued for precision (38).
+     */
     private boolean calcBigDecimalPrecision = SQLServerDriverBooleanProperty.CALC_BIG_DECIMAL_PRECISION
             .getDefaultValue();
 
@@ -3207,9 +3210,15 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
             state = State.OPENED;
 
+            // Socket timeout is bounded by loginTimeout during the login phase.
+            // Reset socket timeout back to the original value.
+            tdsChannel.resetTcpSocketTimeout();
+
             if (connectionlogger.isLoggable(Level.FINER)) {
                 connectionlogger.finer(toString() + " End of connect");
             }
+        } catch (SocketException e) {
+            throw new SQLServerException(e.getMessage(), null);
         } finally {
             // once we exit the connect function, the connection can be only in one of two
             // states, Opened or Closed(if an exception occurred)
@@ -3222,6 +3231,21 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
         return this;
     }
 
+    // log open connection failures
+    private void logConnectFailure(int attemptNumber, SQLServerException e, SQLServerError sqlServerError) {
+        loggerResiliency.finer(toString() + " Connection open - connection failed on attempt: " + attemptNumber + ".");
+
+        if (e != null) {
+            loggerResiliency.finer(
+                    toString() + " Connection open - connection failure. Driver error code: " + e.getDriverErrorCode());
+        }
+
+        if (null != sqlServerError && !sqlServerError.getErrorMessage().isEmpty()) {
+            loggerResiliency.finer(toString() + " Connection open - connection failure. SQL Server error : "
+                    + sqlServerError.getErrorMessage());
+        }
+    }
+
     /**
      * This function is used by non failover and failover cases. Even when we make a standard connection the server can
      * provide us with its FO partner. If no FO information is available a standard connection is made. If the server
@@ -3237,6 +3261,7 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
         int fedauthRetryInterval = BACKOFF_INTERVAL; // milliseconds to sleep (back off) between attempts.
 
         long timeoutUnitInterval;
+        long timeForFirstTry = 0; // time it took to do 1st try in ms
 
         boolean useFailoverHost = false;
         FailoverInfo tempFailover = null;
@@ -3434,34 +3459,24 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
                                     + connectRetryCount + " reached.");
                 }
 
-                int errorCode = e.getErrorCode();
-                int driverErrorCode = e.getDriverErrorCode();
+                // estimate time it took to do 1 try
+                if (attemptNumber == 0) {
+                    timeForFirstTry = (System.currentTimeMillis() - timerStart);
+                }
+
                 sqlServerError = e.getSQLServerError();
 
-                if (SQLServerException.LOGON_FAILED == errorCode // logon failed, ie bad password
-                        || SQLServerException.PASSWORD_EXPIRED == errorCode // password expired
-                        || SQLServerException.USER_ACCOUNT_LOCKED == errorCode // user account locked
-                        || SQLServerException.DRIVER_ERROR_INVALID_TDS == driverErrorCode // invalid TDS
-                        || SQLServerException.DRIVER_ERROR_SSL_FAILED == driverErrorCode // SSL failure
-                        || SQLServerException.DRIVER_ERROR_INTERMITTENT_TLS_FAILED == driverErrorCode // TLS1.2 failure
-                        || SQLServerException.DRIVER_ERROR_UNSUPPORTED_CONFIG == driverErrorCode // unsupported config
-                                                                                                 // (eg Sphinx, invalid
-                                                                                                 // packetsize, etc)
-                        || (SQLServerException.ERROR_SOCKET_TIMEOUT == driverErrorCode // socket timeout
-                                && (!isDBMirroring || attemptNumber > 0)) // If mirroring, only close after failover has been tried (attempt >= 1)
-                        || timerHasExpired(timerExpire)
-                // for non-dbmirroring cases, do not retry after tcp socket connection succeeds
+                if (isFatalError(e) // do not retry on fatal errors
+                        || timerHasExpired(timerExpire) // no time left
+                        || (timerRemaining(timerExpire) < TimeUnit.SECONDS.toMillis(connectRetryInterval)
+                                + 2 * timeForFirstTry) // not enough time for another retry
+                        || (connectRetryCount == 0 && !isDBMirroring && !useTnir) // retries disabled
+                        // retry at least once for TNIR and failover
+                        || (connectRetryCount == 0 && (isDBMirroring || useTnir) && attemptNumber > 0)
+                        || (connectRetryCount != 0 && attemptNumber >= connectRetryCount) // no retries left
                 ) {
                     if (loggerResiliency.isLoggable(Level.FINER)) {
-                        loggerResiliency.finer(
-                                toString() + " Connection open - connection failed on attempt: " + attemptNumber + ".");
-                        loggerResiliency.finer(toString() + " Connection open - connection failure. Driver error code: "
-                                + driverErrorCode);
-                        if (null != sqlServerError && !sqlServerError.getErrorMessage().isEmpty()) {
-                            loggerResiliency
-                                    .finer(toString() + " Connection open - connection failure. SQL Server error : "
-                                            + sqlServerError.getErrorMessage());
-                        }
+                        logConnectFailure(attemptNumber, e, sqlServerError);
                     }
 
                     // close the connection and throw the error back
@@ -3469,15 +3484,7 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
                     throw e;
                 } else {
                     if (loggerResiliency.isLoggable(Level.FINER)) {
-                        loggerResiliency.finer(
-                                toString() + " Connection open - connection failed on attempt: " + attemptNumber + ".");
-                        loggerResiliency.finer(toString() + " Connection open - connection failure. Driver error code: "
-                                + driverErrorCode);
-                        if (null != sqlServerError && !sqlServerError.getErrorMessage().isEmpty()) {
-                            loggerResiliency
-                                    .finer(toString() + " Connection open - connection failure. SQL Server error : "
-                                            + sqlServerError.getErrorMessage());
-                        }
+                        logConnectFailure(attemptNumber, e, sqlServerError);
                     }
 
                     // Close the TDS channel from the failed connection attempt so that we don't
@@ -3496,15 +3503,7 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
                     if (remainingMilliseconds <= fedauthRetryInterval) {
 
                         if (loggerResiliency.isLoggable(Level.FINER)) {
-                            loggerResiliency.finer(toString() + " Connection open - connection failed on attempt: "
-                                    + attemptNumber + ".");
-                            loggerResiliency.finer(toString()
-                                    + " Connection open - connection failure. Driver error code: " + driverErrorCode);
-                            if (null != sqlServerError && !sqlServerError.getErrorMessage().isEmpty()) {
-                                loggerResiliency
-                                        .finer(toString() + " Connection open - connection failure. SQL Server error : "
-                                                + sqlServerError.getErrorMessage());
-                            }
+                            logConnectFailure(attemptNumber, e, sqlServerError);
                         }
 
                         throw e;
@@ -3517,19 +3516,22 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
             // the network with requests, then update sleep interval for next iteration (max 1 second interval)
             // We have to sleep for every attempt in case of non-dbMirroring scenarios (including multisubnetfailover),
             // Whereas for dbMirroring, we sleep for every two attempts as each attempt is to a different server.
-            if (!isDBMirroring || (1 == attemptNumber % 2)) {
-                if (loggerResiliency.isLoggable(Level.FINER)) {
-                    loggerResiliency.finer(toString() + " Connection open - sleeping milisec: " + connectRetryInterval);
-                }
-                if (loggerResiliency.isLoggable(Level.FINER)) {
-                    loggerResiliency.finer(toString() + " Connection open - connection failed on transient error "
-                            + (sqlServerError != null ? sqlServerError.getErrorNumber() : "")
-                            + ". Wait for connectRetryInterval(" + connectRetryInterval + ")s before retry #"
-                            + attemptNumber);
-                }
+            // Make sure there's enough time to do another retry
+            if (!isDBMirroring || (isDBMirroring && (0 == attemptNumber % 2))
+                    && (attemptNumber < connectRetryCount && connectRetryCount != 0) && timerRemaining(
+                            timerExpire) > (TimeUnit.SECONDS.toMillis(connectRetryInterval) + 2 * timeForFirstTry)) {
+
+                // don't wait for TNIR
+                if (!(useTnir && attemptNumber == 0)) {
+                    if (loggerResiliency.isLoggable(Level.FINER)) {
+                        loggerResiliency.finer(toString() + " Connection open - connection failed on transient error "
+                                + (sqlServerError != null ? sqlServerError.getErrorNumber() : "")
+                                + ". Wait for connectRetryInterval(" + connectRetryInterval + ")s before retry #"
+                                + attemptNumber);
+                    }
 
-                sleepForInterval(fedauthRetryInterval);
-                fedauthRetryInterval = (fedauthRetryInterval < 500) ? fedauthRetryInterval * 2 : 1000;
+                    sleepForInterval(TimeUnit.SECONDS.toMillis(connectRetryInterval));
+                }
             }
 
             // Update timeout interval (but no more than the point where we're supposed to fail: timerExpire)
@@ -3622,31 +3624,22 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
         }
     }
 
+    // non recoverable or retryable fatal errors
     boolean isFatalError(SQLServerException e) {
         /*
          * NOTE: If these conditions are modified, consider modification to conditions in SQLServerConnection::login()
          * and Reconnect::run()
          */
+        int errorCode = e.getErrorCode();
+        int driverErrorCode = e.getDriverErrorCode();
 
-        // actual logon failed (e.g. bad password)
-        if ((SQLServerException.LOGON_FAILED == e.getErrorCode())
-                // actual logon failed (e.g. password expired)
-                || (SQLServerException.PASSWORD_EXPIRED == e.getErrorCode())
-                // actual logon failed (e.g. user account locked)
-                || (SQLServerException.USER_ACCOUNT_LOCKED == e.getErrorCode())
-                // invalid TDS received from server
-                || (SQLServerException.DRIVER_ERROR_INVALID_TDS == e.getDriverErrorCode())
-                // failure negotiating SSL
-                || (SQLServerException.DRIVER_ERROR_SSL_FAILED == e.getDriverErrorCode())
-                // failure TLS1.2
-                || (SQLServerException.DRIVER_ERROR_INTERMITTENT_TLS_FAILED == e.getDriverErrorCode())
-                // unsupported configuration (e.g. Sphinx, invalid packet size, etc.)
-                || (SQLServerException.DRIVER_ERROR_UNSUPPORTED_CONFIG == e.getDriverErrorCode())
-                // no more time to try again
-                || (SQLServerException.ERROR_SOCKET_TIMEOUT == e.getDriverErrorCode()))
-            return true;
-        else
-            return false;
+        return ((SQLServerException.LOGON_FAILED == errorCode) // logon failed (eg bad password)
+                || (SQLServerException.PASSWORD_EXPIRED == errorCode) // password expired
+                || (SQLServerException.USER_ACCOUNT_LOCKED == errorCode) // user account locked
+                || (SQLServerException.DRIVER_ERROR_INVALID_TDS == driverErrorCode) // invalid TDS from server
+                || (SQLServerException.DRIVER_ERROR_SSL_FAILED == driverErrorCode) // SSL failure
+                || (SQLServerException.DRIVER_ERROR_INTERMITTENT_TLS_FAILED == driverErrorCode) // TLS1.2 failure
+                || (SQLServerException.DRIVER_ERROR_UNSUPPORTED_CONFIG == driverErrorCode)); // unsupported config (eg Sphinx, invalid packet size ,etc)
     }
 
     // reset all params that could have been changed due to ENVCHANGE tokens to defaults,
@@ -3707,7 +3700,7 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
     }
 
     /**
-     * Get time remaining to timer expiry
+     * Get time remaining to timer expiry (in ms)
      * 
      * @param timerExpire
      * @return remaining time to expiry
@@ -4798,6 +4791,8 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
         // Clean-up queue etc. related to batching of prepared statement discard actions (sp_unprepare).
         cleanupPreparedStatementDiscardActions();
+
+        ActivityCorrelator.cleanupActivityId();
     }
 
     /**
@@ -4819,6 +4814,8 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
             }
             notifyPooledConnection(null);
 
+            ActivityCorrelator.cleanupActivityId();
+
             if (connectionlogger.isLoggable(Level.FINER)) {
                 connectionlogger.finer(toString() + " Connection closed and returned to connection pool");
             }
@@ -4943,6 +4940,27 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
         }
     }
 
+    // Any changes to SQLWarnings should be synchronized.
+    /** Used to add plain SQLWarning messages (if they do not hold extended information, like: ErrorSeverity, ServerName, ProcName etc */
+    void addWarning(SQLWarning sqlWarning) {
+        warningSynchronization.lock();
+        try {
+            if (null == sqlWarnings) {
+                sqlWarnings = sqlWarning;
+            } else {
+                sqlWarnings.setNextWarning(sqlWarning);
+            }
+        } finally {
+            warningSynchronization.unlock();
+        }
+    }
+
+    // Any changes to SQLWarnings should be synchronized.
+    /** Used to add messages that holds extended information, like: ErrorSeverity, ServerName, ProcName etc */
+    void addWarning(ISQLServerMessage sqlServerMessage) {
+        addWarning(new SQLServerWarning(sqlServerMessage.getSQLServerMessage()));
+    }
+
     @Override
     public void clearWarnings() throws SQLServerException {
         warningSynchronization.lock();
@@ -5979,7 +5997,7 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
         String user = activeConnectionProperties.getProperty(SQLServerDriverStringProperty.USER.toString());
 
         // No of milliseconds to sleep for the initial back off.
-        int sleepInterval = BACKOFF_INTERVAL;
+        int fedauthSleepInterval = BACKOFF_INTERVAL;
 
         if (!msalContextExists()
                 && !authenticationString.equalsIgnoreCase(SqlAuthentication.ACTIVE_DIRECTORY_INTEGRATED.toString())) {
@@ -6078,7 +6096,7 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
                         int millisecondsRemaining = timerRemaining(timerExpire);
                         if (ActiveDirectoryAuthentication.GET_ACCESS_TOKEN_TRANSIENT_ERROR != errorCategory
-                                || timerHasExpired(timerExpire) || (sleepInterval >= millisecondsRemaining)) {
+                                || timerHasExpired(timerExpire) || (fedauthSleepInterval >= millisecondsRemaining)) {
 
                             String errorStatus = Integer.toHexString(adalException.getStatus());
 
@@ -6102,13 +6120,14 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
 
                         if (connectionlogger.isLoggable(Level.FINER)) {
                             connectionlogger.fine(toString() + " SQLServerConnection.getFedAuthToken sleeping: "
-                                    + sleepInterval + " milliseconds.");
+                                    + fedauthSleepInterval + " milliseconds.");
                             connectionlogger.fine(toString() + " SQLServerConnection.getFedAuthToken remaining: "
                                     + millisecondsRemaining + " milliseconds.");
                         }
 
-                        sleepForInterval(sleepInterval);
-                        sleepInterval = sleepInterval * 2;
+                        sleepForInterval(fedauthSleepInterval);
+                        fedauthSleepInterval = (fedauthSleepInterval < 500) ? fedauthSleepInterval * 2 : 1000;
+
                     }
                 }
                 // else choose MSAL4J for integrated authentication. This option is supported for both windows and unix,
@@ -8514,6 +8533,31 @@ public class SQLServerConnection implements ISQLServerConnection, java.io.Serial
     public String getIPAddressPreference() {
         return activeConnectionProperties.getProperty(SQLServerDriverStringProperty.IPADDRESS_PREFERENCE.toString());
     }
+
+    /** Message handler */
+    private transient ISQLServerMessageHandler serverMessageHandler;
+
+    /**
+     * Set current message handler
+     * 
+     * @param messageHandler
+     *        message handler
+     * @return The previously installed message handler (null if none)
+     */
+    @Override
+    public ISQLServerMessageHandler setServerMessageHandler(ISQLServerMessageHandler messageHandler) {
+        ISQLServerMessageHandler installedMessageHandler = this.serverMessageHandler;
+        this.serverMessageHandler = messageHandler;
+        return installedMessageHandler;
+    }
+
+    /**
+     * @return Get Currently installed message handler on the connection
+     */
+    @Override
+    public ISQLServerMessageHandler getServerMessageHandler() {
+        return this.serverMessageHandler;
+    }
 }
 
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionPoolProxy.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionPoolProxy.java
index ffda2dd6..5ed4b7b0 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionPoolProxy.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionPoolProxy.java
@@ -692,6 +692,16 @@ class SQLServerConnectionPoolProxy implements ISQLServerConnection, java.io.Seri
         wrappedConnection.setAccessTokenCallbackClass(accessTokenCallbackClass);
     }
 
+    @Override
+    public ISQLServerMessageHandler getServerMessageHandler() {
+        return wrappedConnection.getServerMessageHandler();
+    }
+    
+    @Override
+    public ISQLServerMessageHandler setServerMessageHandler(ISQLServerMessageHandler messageHandler) {
+    	return wrappedConnection.setServerMessageHandler(messageHandler);
+    }
+
     /**
      * Returns the current value for 'calcBigDecimalPrecision'.
      *
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerDatabaseMetaData.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerDatabaseMetaData.java
index 1d286887..5c23a477 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerDatabaseMetaData.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerDatabaseMetaData.java
@@ -258,11 +258,20 @@ public final class SQLServerDatabaseMetaData implements java.sql.DatabaseMetaDat
     private static final String IS_AUTOINCREMENT = "IS_AUTOINCREMENT";
     private static final String ACTIVITY_ID = " ActivityId: ";
 
+    private static final String NVARCHAR = JDBCType.NVARCHAR.name();
+    private static final String VARCHAR = JDBCType.VARCHAR.name();
+    private static final String INTEGER = JDBCType.INTEGER.name();
+    private static final String SMALLINT = JDBCType.SMALLINT.name();
+
     private static final String SQL_KEYWORDS = createSqlKeyWords();
 
     // Use LinkedHashMap to force retrieve elements in order they were inserted
     /** getColumns columns */
     private LinkedHashMap<Integer, String> getColumnsDWColumns = null;
+
+    /** getTypes columns */
+    private LinkedHashMap<Integer, String> getTypesDWColumns = null;
+
     /** getImportedKeys columns */
     private volatile LinkedHashMap<Integer, String> getImportedKeysDWColumns;
     private static final Lock LOCK = new ReentrantLock();
@@ -630,10 +639,13 @@ public final class SQLServerDatabaseMetaData implements java.sql.DatabaseMetaDat
 
                     + "INSERT INTO @mssqljdbc_temp_sp_columns_result EXEC sp_columns_100 ?,?,?,?,?,?;"
 
-                    + "SELECT TABLE_QUALIFIER AS TABLE_CAT, TABLE_OWNER AS TABLE_SCHEM, TABLE_NAME, COLUMN_NAME, DATA_TYPE,"
-                    + "TYPE_NAME, PRECISION AS COLUMN_SIZE, LENGTH AS BUFFER_LENGTH, SCALE AS DECIMAL_DIGITS, RADIX AS NUM_PREC_RADIX,"
-                    + "NULLABLE, REMARKS, COLUMN_DEF, SQL_DATA_TYPE, SQL_DATETIME_SUB, CHAR_OCTET_LENGTH, ORDINAL_POSITION, IS_NULLABLE,"
-                    + "NULL AS SCOPE_CATALOG, NULL AS SCOPE_SCHEMA, NULL AS SCOPE_TABLE, SS_DATA_TYPE AS SOURCE_DATA_TYPE,"
+                    + "SELECT TABLE_QUALIFIER AS TABLE_CAT, TABLE_OWNER AS TABLE_SCHEM, TABLE_NAME, COLUMN_NAME, "
+                    + "CAST(DATA_TYPE AS INT) AS DATA_TYPE,TYPE_NAME, PRECISION AS COLUMN_SIZE, LENGTH AS BUFFER_LENGTH, "
+                    + "CAST(SCALE AS INT) AS DECIMAL_DIGITS, CAST(RADIX AS INT) AS NUM_PREC_RADIX,CAST(NULLABLE AS INT) AS NULLABLE, "
+                    + "CAST(REMARKS AS VARCHAR) AS REMARKS, COLUMN_DEF, CAST(SQL_DATA_TYPE AS INT) AS SQL_DATA_TYPE, "
+                    + "CAST(SQL_DATETIME_SUB AS INT) AS SQL_DATETIME_SUB, CHAR_OCTET_LENGTH, ORDINAL_POSITION, IS_NULLABLE,"
+                    + "CAST(NULL AS VARCHAR) AS SCOPE_CATALOG, CAST(NULL AS VARCHAR) AS SCOPE_SCHEMA, CAST(NULL AS VARCHAR) AS SCOPE_TABLE, "
+                    + "CAST(SS_DATA_TYPE AS SMALLINT) AS SOURCE_DATA_TYPE, "
                     + "CASE SS_IS_IDENTITY WHEN 0 THEN 'NO' WHEN 1 THEN 'YES' WHEN '' THEN '' END AS IS_AUTOINCREMENT,"
                     + "CASE SS_IS_COMPUTED WHEN 0 THEN 'NO' WHEN 1 THEN 'YES' WHEN '' THEN '' END AS IS_GENERATEDCOLUMN, "
                     + "SS_IS_SPARSE, SS_IS_COLUMN_SET, SS_UDT_CATALOG_NAME, SS_UDT_SCHEMA_NAME, SS_UDT_ASSEMBLY_TYPE_NAME,"
@@ -721,6 +733,53 @@ public final class SQLServerDatabaseMetaData implements java.sql.DatabaseMetaDat
                     getColumnsDWColumns.put(27, SS_XML_SCHEMACOLLECTION_SCHEMA_NAME);
                     getColumnsDWColumns.put(28, SS_XML_SCHEMACOLLECTION_NAME);
                 }
+                if (null == getTypesDWColumns) {
+                    getTypesDWColumns = new LinkedHashMap<>();
+                    getTypesDWColumns.put(1, NVARCHAR); // TABLE_CAT
+                    getTypesDWColumns.put(2, NVARCHAR); // TABLE_SCHEM
+                    getTypesDWColumns.put(3, NVARCHAR); // TABLE_NAME
+                    getTypesDWColumns.put(4, NVARCHAR); // COLUMN_NAME
+                    getTypesDWColumns.put(5, INTEGER); // DATA_TYPE
+                    getTypesDWColumns.put(6, NVARCHAR); // TYPE_NAME
+                    getTypesDWColumns.put(7, INTEGER); // COLUMN_SIZE
+                    getTypesDWColumns.put(8, INTEGER); // BUFFER_LENGTH
+                    getTypesDWColumns.put(9, INTEGER); // DECIMAL_DIGITS
+                    getTypesDWColumns.put(10, INTEGER); // NUM_PREC_RADIX
+                    getTypesDWColumns.put(11, INTEGER); // NULLABLE
+                    getTypesDWColumns.put(12, VARCHAR); // REMARKS
+                    getTypesDWColumns.put(13, NVARCHAR); // COLUMN_DEF
+                    getTypesDWColumns.put(14, INTEGER); // SQL_DATA_TYPE
+                    getTypesDWColumns.put(15, INTEGER); // SQL_DATETIME_SUB
+                    getTypesDWColumns.put(16, INTEGER); // CHAR_OCTET_LENGTH
+                    getTypesDWColumns.put(17, INTEGER); // ORDINAL_POSITION
+                    getTypesDWColumns.put(18, VARCHAR); // IS_NULLABLE
+                    /*
+                     * Use negative value keys to indicate that this column doesn't exist in SQL Server and should just
+                     * be queried as 'NULL'
+                     */
+                    getTypesDWColumns.put(-1, VARCHAR); // SCOPE_CATALOG
+                    getTypesDWColumns.put(-2, VARCHAR); // SCOPE_SCHEMA
+                    getTypesDWColumns.put(-3, VARCHAR); // SCOPE_TABLE
+                    getTypesDWColumns.put(29, SMALLINT); // SOURCE_DATA_TYPE
+                    getTypesDWColumns.put(22, VARCHAR); // IS_AUTOINCREMENT
+                    getTypesDWColumns.put(21, VARCHAR); // IS_GENERATEDCOLUMN
+                    getTypesDWColumns.put(19, SMALLINT); // SS_IS_SPARSE
+                    getTypesDWColumns.put(20, SMALLINT); // SS_IS_COLUMN_SET
+                    getTypesDWColumns.put(23, NVARCHAR); // SS_UDT_CATALOG_NAME
+                    getTypesDWColumns.put(24, NVARCHAR); // SS_UDT_SCHEMA_NAME
+                    getTypesDWColumns.put(25, NVARCHAR); // SS_UDT_ASSEMBLY_TYPE_NAME
+                    getTypesDWColumns.put(26, NVARCHAR); // SS_XML_SCHEMACOLLECTION_CATALOG_NAME
+                    getTypesDWColumns.put(27, NVARCHAR); // SS_XML_SCHEMACOLLECTION_SCHEMA_NAME
+                    getTypesDWColumns.put(28, NVARCHAR); // SS_XML_SCHEMACOLLECTION_NAME
+                }
+
+                // Ensure there is a data type for every metadata column
+                if (getColumnsDWColumns.size() != getTypesDWColumns.size()) {
+                    MessageFormat form = new MessageFormat(
+                            SQLServerException.getErrString("R_colCountNotMatchColTypeCount"));
+                    Object[] msgArgs = {getColumnsDWColumns.size(), getTypesDWColumns.size()};
+                    throw new IllegalArgumentException(form.format(msgArgs));
+                }
             } finally {
                 LOCK.unlock();
             }
@@ -744,7 +803,7 @@ public final class SQLServerDatabaseMetaData implements java.sql.DatabaseMetaDat
                         if (!isFirstRow) {
                             azureDwSelectBuilder.append(" UNION ALL ");
                         }
-                        azureDwSelectBuilder.append(generateAzureDWSelect(rs, getColumnsDWColumns));
+                        azureDwSelectBuilder.append(generateAzureDWSelect(rs, getColumnsDWColumns, getTypesDWColumns));
                         isFirstRow = false;
                     }
 
@@ -780,28 +839,41 @@ public final class SQLServerDatabaseMetaData implements java.sql.DatabaseMetaDat
         }
     }
 
-    private String generateAzureDWSelect(ResultSet rs, Map<Integer, String> columns) throws SQLException {
+    private String generateAzureDWSelect(ResultSet rs, Map<Integer, String> columns,
+            Map<Integer, String> types) throws SQLException {
         StringBuilder sb = new StringBuilder("SELECT ");
+
         for (Entry<Integer, String> p : columns.entrySet()) {
+            String dataType = types.get(p.getKey());
+
+            // Verify there is a valid column entry in the Data Type lookup map
+            if (dataType == null) {
+                MessageFormat form = new MessageFormat(SQLServerException.getErrString("R_invalidArgument"));
+                Object[] msgArgs = {p.getKey()};
+                throw new SQLServerException(null, form.format(msgArgs), null, 0, true);
+            }
+
+            sb.append("CAST(");
             if (p.getKey() < 0) {
-                sb.append("NULL");
+                sb.append("NULL AS " + dataType);
             } else {
                 Object o = rs.getObject(p.getKey());
                 if (null == o) {
-                    sb.append("NULL");
+                    sb.append("NULL AS " + dataType);
                 } else if (o instanceof Number) {
                     if (IS_AUTOINCREMENT.equalsIgnoreCase(p.getValue())
                             || IS_GENERATEDCOLUMN.equalsIgnoreCase(p.getValue())) {
                         sb.append("'").append(Util.escapeSingleQuotes(Util.zeroOneToYesNo(((Number) o).intValue())))
-                                .append("'");
+                                .append("' AS ").append(dataType);
                     } else {
-                        sb.append(o.toString());
+                        sb.append(o.toString()).append(" AS ").append(dataType);
                     }
                 } else {
-                    sb.append("'").append(Util.escapeSingleQuotes(o.toString())).append("'");
+                    sb.append("'").append(Util.escapeSingleQuotes(o.toString())).append("' AS ").append(dataType)
+                            .append("(").append(Integer.toString(o.toString().length())).append(")");
                 }
             }
-            sb.append(" AS ").append(p.getValue()).append(",");
+            sb.append(") AS ").append(p.getValue()).append(",");
         }
         sb.setLength(sb.length() - 1);
         return sb.toString();
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java
index 3454e5ae..b3e96904 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerError.java
@@ -6,12 +6,15 @@
 package com.microsoft.sqlserver.jdbc;
 
 import java.io.Serializable;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
 
 
 /**
  * SQLServerError represents a TDS error or message event.
  */
-public final class SQLServerError extends StreamPacket implements Serializable {
+public final class SQLServerError extends StreamPacket implements Serializable, ISQLServerMessage {
 
     /**
      * List SQL Server transient errors drivers will retry on from
@@ -221,6 +224,18 @@ public final class SQLServerError extends StreamPacket implements Serializable {
         super(TDS.TDS_ERR);
     }
 
+    SQLServerError(SQLServerError errorMsg) {
+        super(TDS.TDS_ERR);
+        this.errorNumber = errorMsg.errorNumber;
+        this.errorState = errorMsg.errorState;
+        this.errorSeverity = errorMsg.errorSeverity;
+        this.errorMessage = errorMsg.errorMessage;
+        this.serverName = errorMsg.serverName;
+        this.procName = errorMsg.procName;
+        this.lineNumber = errorMsg.lineNumber;
+    }
+
+    @Override
     void setFromTDS(TDSReader tdsReader) throws SQLServerException {
         if (TDS.TDS_ERR != tdsReader.readUnsignedByte())
             assert false;
@@ -237,4 +252,103 @@ public final class SQLServerError extends StreamPacket implements Serializable {
         procName = tdsReader.readUnicodeString(tdsReader.readUnsignedByte());
         lineNumber = tdsReader.readUnsignedInt();
     }
+
+    /**
+     * Holds any "overflow messages", or messages that has been added <b>after</b> the first message.
+     * <p>
+     * This is later on used when creating a SQLServerException.<br>
+     * Where all entries in the errorChain will be added {@link java.sql.SQLException#setNextException(SQLException)}
+     */
+    private List<SQLServerError> errorChain;
+
+    void addError(SQLServerError sqlServerError) {
+        if (errorChain == null) {
+            errorChain = new ArrayList<>();
+        }
+        errorChain.add(sqlServerError);
+    }
+
+    List<SQLServerError> getErrorChain() {
+        return errorChain;
+    }
+
+    @Override
+    public SQLServerError getSQLServerMessage() {
+        return this;
+    }
+
+    /**
+     * Downgrade a Error message into a Info message
+     * <p>
+     * This simply create a SQLServerInfoMessage from this SQLServerError,
+     * without changing the message content.
+     * 
+     * @return ISQLServerMessage
+     */
+    public ISQLServerMessage toSQLServerInfoMessage() {
+        return toSQLServerInfoMessage(-1, -1);
+    }
+
+    /**
+     * Downgrade a Error message into a Info message
+     * <p>
+     * This simply create a SQLServerInfoMessage from this SQLServerError,
+     * 
+     * @param newErrorSeverity
+     *        - The new ErrorSeverity
+     * 
+     * @return ISQLServerMessage
+     */
+    public ISQLServerMessage toSQLServerInfoMessage(int newErrorSeverity) {
+        return toSQLServerInfoMessage(newErrorSeverity, -1);
+    }
+
+    /**
+     * Downgrade a Error message into a Info message
+     * <p>
+     * This simply create a SQLServerInfoMessage from this SQLServerError,
+     * 
+     * @param newErrorSeverity
+     *        - If you want to change the ErrorSeverity (-1: leave unchanged)
+     * @param newErrorNumber
+     *        - If you want to change the ErrorNumber (-1: leave unchanged)
+     * 
+     * @return ISQLServerMessage
+     */
+    public ISQLServerMessage toSQLServerInfoMessage(int newErrorSeverity, int newErrorNumber) {
+        if (newErrorSeverity != -1) {
+            this.setErrorSeverity(newErrorSeverity);
+        }
+
+        if (newErrorNumber != -1) {
+            this.setErrorNumber(newErrorNumber);
+        }
+
+        return new SQLServerInfoMessage(this);
+    }
+
+    /**
+     * Set a new ErrorSeverity for this Message
+     * 
+     * @param newSeverity
+     *        new severity
+     */
+    public void setErrorSeverity(int newSeverity) {
+        this.errorSeverity = newSeverity;
+    }
+
+    /**
+     * Set a new ErrorNumber for this Message
+     * 
+     * @param newErrorNumber
+     *        new error number
+     */
+    public void setErrorNumber(int newErrorNumber) {
+        this.errorNumber = newErrorNumber;
+    }
+
+    @Override
+    public SQLException toSqlExceptionOrSqlWarning() {
+        return new SQLServerException(this);
+    }
 }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerException.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerException.java
index 7d4fa23f..77367708 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerException.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerException.java
@@ -7,6 +7,7 @@ package com.microsoft.sqlserver.jdbc;
 
 import java.sql.SQLFeatureNotSupportedException;
 import java.text.MessageFormat;
+import java.util.List;
 import java.util.UUID;
 import java.util.logging.Level;
 
@@ -178,6 +179,15 @@ public final class SQLServerException extends java.sql.SQLException {
         logException(obj, errText, bStack);
     }
 
+    public SQLServerException(SQLServerError sqlServerError) {
+        super(sqlServerError.getErrorMessage(), 
+            generateStateCode(null, sqlServerError.getErrorNumber(), sqlServerError.getErrorState()), 
+            sqlServerError.getErrorNumber(), 
+            null);
+
+        this.sqlServerError = sqlServerError;
+	}
+
     /**
      * Constructs a new SQLServerException.
      * 
@@ -261,6 +271,21 @@ public final class SQLServerException extends java.sql.SQLException {
                 SQLServerException.checkAndAppendClientConnId(errText, con), state, sqlServerError, bStack);
         theException.setDriverErrorCode(DRIVER_ERROR_FROM_DATABASE);
 
+        // Add any extra messages to the SQLException error chain
+        List<SQLServerError> errorChain = sqlServerError.getErrorChain();
+        if (errorChain != null) {
+            for (SQLServerError srvError : errorChain) {
+                String state2 = generateStateCode(con, srvError.getErrorNumber(), srvError.getErrorState());
+
+                SQLServerException chainException = new SQLServerException(obj,
+                        SQLServerException.checkAndAppendClientConnId(srvError.getErrorMessage(), con), 
+                        state2, srvError, bStack);
+                chainException.setDriverErrorCode(DRIVER_ERROR_FROM_DATABASE);
+                
+                theException.setNextException(chainException);
+            }
+        }
+
         // Close the connection if we get a severity 20 or higher error class (nClass is severity of error).
         if ((sqlServerError.getErrorSeverity() >= 20) && (null != con)) {
             con.notifyPooledConnection(theException);
@@ -395,16 +420,17 @@ public final class SQLServerException extends java.sql.SQLException {
     static String checkAndAppendClientConnId(String errMsg, SQLServerConnection conn) {
         if (null != conn && conn.isConnected()) {
             UUID clientConnId = conn.getClientConIdInternal();
-            assert null != clientConnId;
-            StringBuilder sb = new StringBuilder(errMsg);
-            // This syntax of adding connection id is matched in a retry logic. If anything changes here, make
-            // necessary changes to enableSSL() function's exception handling mechanism.
-            sb.append(LOG_CLIENT_CONNECTION_ID_PREFIX);
-            sb.append(clientConnId.toString());
-            return sb.toString();
-        } else {
-            return errMsg;
+            if (null != clientConnId) {
+                StringBuilder sb = (errMsg != null) ? new StringBuilder(errMsg) : new StringBuilder();
+                // This syntax of adding connection id is matched in a retry logic. If anything changes here, make
+                // necessary changes to enableSSL() function's exception handling mechanism.
+                sb.append(LOG_CLIENT_CONNECTION_ID_PREFIX);
+                sb.append(clientConnId.toString());
+                return sb.toString();
+            }
         }
+        return (errMsg != null) ? errMsg : "";
+
     }
 
     static void throwNotSupportedException(SQLServerConnection con, Object obj) throws SQLServerException {
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerInfoMessage.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerInfoMessage.java
new file mode 100644
index 00000000..0b3a6075
--- /dev/null
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerInfoMessage.java
@@ -0,0 +1,135 @@
+/*
+ * 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.sql.SQLException;
+
+
+/**
+ * Holds information about SQL Server messages that is considered as Informational Messages (normally if SQL Server Severity is at 10)
+ * <p>
+ * Instead of just holding the SQL Server message (like a normal SQLWarning, it also holds all the
+ * SQL Servers extended information, like: ErrorSeverity, ServerName, ProcName etc
+ * <p>
+ * This enables client to print out extra information about the message.<br>
+ * Like: In what procedure was the message produced.
+ * <p>
+ * A SQLServerInfoMessage is produced when reading the TDS Stream and added to the Connection as a SQLServerWarning
+ */
+public final class SQLServerInfoMessage extends StreamPacket implements ISQLServerMessage {
+    SQLServerError msg = new SQLServerError();
+
+    SQLServerInfoMessage() {
+        super(TDS.TDS_MSG);
+    }
+
+    SQLServerInfoMessage(SQLServerError msg) {
+        super(TDS.TDS_MSG);
+        this.msg = msg;
+    }
+
+    @Override
+    void setFromTDS(TDSReader tdsReader) throws SQLServerException {
+        if (TDS.TDS_MSG != tdsReader.readUnsignedByte())
+            assert false;
+        msg.setContentsFromTDS(tdsReader);
+    }
+
+    @Override
+    public SQLServerError getSQLServerMessage() {
+        return msg;
+    }
+
+    @Override
+    public String getErrorMessage() {
+        return msg.getErrorMessage();
+    }
+
+    @Override
+    public int getErrorNumber() {
+        return msg.getErrorNumber();
+    }
+
+    @Override
+    public int getErrorState() {
+        return msg.getErrorState();
+    }
+
+    @Override
+    public int getErrorSeverity() {
+        return msg.getErrorSeverity();
+    }
+
+    @Override
+    public String getServerName() {
+        return msg.getServerName();
+    }
+
+    @Override
+    public String getProcedureName() {
+        return msg.getProcedureName();
+    }
+
+    @Override
+    public long getLineNumber() {
+        return msg.getLineNumber();
+    }
+
+    /**
+     * Upgrade a Info message into a Error message
+     * <p>
+     * This simply create a SQLServerError from this SQLServerInfoMessage,
+     * without changing the message content.
+     * 
+     * @return ISQLServerMessage
+     */
+    public ISQLServerMessage toSQLServerError() {
+        return toSQLServerError(-1, -1);
+    }
+
+    /**
+     * Upgrade a Info message into a Error message
+     * <p>
+     * This simply create a SQLServerError from this SQLServerInfoMessage.
+     * 
+     * @param newErrorSeverity
+     *        - The new ErrorSeverity
+     * 
+     * @return ISQLServerMessage
+     */
+    public ISQLServerMessage toSQLServerError(int newErrorSeverity) {
+        return toSQLServerError(newErrorSeverity, -1);
+    }
+
+    /**
+     * Upgrade a Info message into a Error message
+     * <p>
+     * This simply create a SQLServerError from this SQLServerInfoMessage.
+     * 
+     * @param newErrorSeverity
+     *        - If you want to change the ErrorSeverity (-1: leave unchanged)
+     * @param newErrorNumber
+     *        - If you want to change the ErrorNumber (-1: leave unchanged)
+     * 
+     * @return ISQLServerMessage
+     */
+    public ISQLServerMessage toSQLServerError(int newErrorSeverity, int newErrorNumber) {
+        if (newErrorSeverity != -1) {
+            this.msg.setErrorSeverity(newErrorSeverity);
+        }
+
+        if (newErrorNumber != -1) {
+            this.msg.setErrorNumber(newErrorNumber);
+        }
+
+        return new SQLServerError(this.msg);
+    }
+
+    @Override
+    public SQLException toSqlExceptionOrSqlWarning() {
+        return new SQLServerWarning(this.msg);
+    }
+}
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerMSAL4JUtils.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerMSAL4JUtils.java
index c8f2fb34..1ae69891 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerMSAL4JUtils.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerMSAL4JUtils.java
@@ -14,6 +14,7 @@ import java.net.MalformedURLException;
 import java.net.URI;
 import java.net.URISyntaxException;
 
+import java.security.MessageDigest;
 import java.text.MessageFormat;
 
 import java.util.Collections;
@@ -21,6 +22,7 @@ import java.util.Enumeration;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -61,6 +63,8 @@ class SQLServerMSAL4JUtils {
     static final String SLASH_DEFAULT = "/.default";
     static final String ACCESS_TOKEN_EXPIRE = "access token expires: ";
 
+    private static final TokenCacheMap TOKEN_CACHE_MAP = new TokenCacheMap();
+
     private final static String LOGCONTEXT = "MSAL version "
             + com.microsoft.aad.msal4j.PublicClientApplication.class.getPackage().getImplementationVersion() + ": ";
 
@@ -84,10 +88,17 @@ class SQLServerMSAL4JUtils {
         lock.lock();
 
         try {
+            String hashedSecret = getHashedSecret(new String[] {fedAuthInfo.stsurl, user, password});
+            PersistentTokenCacheAccessAspect persistentTokenCacheAccessAspect = TOKEN_CACHE_MAP.getEntry(hashedSecret);
+
+            if (null == persistentTokenCacheAccessAspect) {
+                persistentTokenCacheAccessAspect = new PersistentTokenCacheAccessAspect();
+                TOKEN_CACHE_MAP.addEntry(hashedSecret, persistentTokenCacheAccessAspect);
+            }
+
             final PublicClientApplication pca = PublicClientApplication
                     .builder(ActiveDirectoryAuthentication.JDBC_FEDAUTH_CLIENT_ID).executorService(executorService)
-                    .setTokenCacheAccessAspect(PersistentTokenCacheAccessAspect.getInstance())
-                    .authority(fedAuthInfo.stsurl).build();
+                    .setTokenCacheAccessAspect(persistentTokenCacheAccessAspect).authority(fedAuthInfo.stsurl).build();
 
             final CompletableFuture<IAuthenticationResult> future = pca.acquireToken(UserNamePasswordParameters
                     .builder(Collections.singleton(fedAuthInfo.spn + SLASH_DEFAULT), user, password.toCharArray())
@@ -132,11 +143,19 @@ class SQLServerMSAL4JUtils {
         lock.lock();
 
         try {
+            String hashedSecret = getHashedSecret(
+                    new String[] {fedAuthInfo.stsurl, aadPrincipalID, aadPrincipalSecret});
+            PersistentTokenCacheAccessAspect persistentTokenCacheAccessAspect = TOKEN_CACHE_MAP.getEntry(hashedSecret);
+
+            if (null == persistentTokenCacheAccessAspect) {
+                persistentTokenCacheAccessAspect = new PersistentTokenCacheAccessAspect();
+                TOKEN_CACHE_MAP.addEntry(hashedSecret, persistentTokenCacheAccessAspect);
+            }
+
             IClientCredential credential = ClientCredentialFactory.createFromSecret(aadPrincipalSecret);
             ConfidentialClientApplication clientApplication = ConfidentialClientApplication
                     .builder(aadPrincipalID, credential).executorService(executorService)
-                    .setTokenCacheAccessAspect(PersistentTokenCacheAccessAspect.getInstance())
-                    .authority(fedAuthInfo.stsurl).build();
+                    .setTokenCacheAccessAspect(persistentTokenCacheAccessAspect).authority(fedAuthInfo.stsurl).build();
 
             final CompletableFuture<IAuthenticationResult> future = clientApplication
                     .acquireToken(ClientCredentialParameters.builder(scopes).build());
@@ -181,6 +200,15 @@ class SQLServerMSAL4JUtils {
         lock.lock();
 
         try {
+            String hashedSecret = getHashedSecret(new String[] {fedAuthInfo.stsurl, aadPrincipalID, certFile,
+                    certPassword, certKey, certKeyPassword});
+            PersistentTokenCacheAccessAspect persistentTokenCacheAccessAspect = TOKEN_CACHE_MAP.getEntry(hashedSecret);
+
+            if (null == persistentTokenCacheAccessAspect) {
+                persistentTokenCacheAccessAspect = new PersistentTokenCacheAccessAspect();
+                TOKEN_CACHE_MAP.addEntry(hashedSecret, persistentTokenCacheAccessAspect);
+            }
+
             ConfidentialClientApplication clientApplication = null;
 
             // check if cert is PKCS12 first
@@ -202,8 +230,7 @@ class SQLServerMSAL4JUtils {
 
                 IClientCredential credential = ClientCredentialFactory.createFromCertificate(is, certPassword);
                 clientApplication = ConfidentialClientApplication.builder(aadPrincipalID, credential)
-                        .executorService(executorService)
-                        .setTokenCacheAccessAspect(PersistentTokenCacheAccessAspect.getInstance())
+                        .executorService(executorService).setTokenCacheAccessAspect(persistentTokenCacheAccessAspect)
                         .authority(fedAuthInfo.stsurl).build();
             } catch (FileNotFoundException e) {
                 // re-throw if file not there no point to try another format
@@ -232,8 +259,7 @@ class SQLServerMSAL4JUtils {
 
                 IClientCredential credential = ClientCredentialFactory.createFromCertificate(privateKey, cert);
                 clientApplication = ConfidentialClientApplication.builder(aadPrincipalID, credential)
-                        .executorService(executorService)
-                        .setTokenCacheAccessAspect(PersistentTokenCacheAccessAspect.getInstance())
+                        .executorService(executorService).setTokenCacheAccessAspect(persistentTokenCacheAccessAspect)
                         .authority(fedAuthInfo.stsurl).build();
             }
 
@@ -449,4 +475,45 @@ class SQLServerMSAL4JUtils {
             return new SQLServerException(form.format(msgArgs), null, 0, correctedExecutionException);
         }
     }
+
+    private static String getHashedSecret(String[] secrets) throws SQLServerException {
+        try {
+            MessageDigest md = MessageDigest.getInstance("SHA-256");
+            for (String secret : secrets) {
+                if (null != secret) {
+                    md.update(secret.getBytes(java.nio.charset.StandardCharsets.UTF_16LE));
+                }
+            }
+            return new String(md.digest());
+        } catch (NoSuchAlgorithmException e) {
+            throw new SQLServerException(SQLServerException.getErrString("R_NoSHA256Algorithm"), e);
+        }
+    }
+
+    private static class TokenCacheMap {
+        private ConcurrentHashMap<String, PersistentTokenCacheAccessAspect> tokenCacheMap = new ConcurrentHashMap<>();
+
+        PersistentTokenCacheAccessAspect getEntry(String key) {
+            PersistentTokenCacheAccessAspect persistentTokenCacheAccessAspect = tokenCacheMap.get(key);
+
+            if (null != persistentTokenCacheAccessAspect) {
+                if (System.currentTimeMillis() > persistentTokenCacheAccessAspect.getExpiryTime()) {
+                    tokenCacheMap.remove(key);
+
+                    persistentTokenCacheAccessAspect = new PersistentTokenCacheAccessAspect();
+                    persistentTokenCacheAccessAspect
+                            .setExpiryTime(System.currentTimeMillis() + PersistentTokenCacheAccessAspect.TIME_TO_LIVE);
+
+                    tokenCacheMap.put(key, persistentTokenCacheAccessAspect);
+                }
+            }
+
+            return persistentTokenCacheAccessAspect;
+        }
+
+        void addEntry(String key, PersistentTokenCacheAccessAspect value) {
+            value.setExpiryTime(System.currentTimeMillis() + PersistentTokenCacheAccessAspect.TIME_TO_LIVE);
+            tokenCacheMap.put(key, value);
+        }
+    }
 }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java
index 7789b372..4274e691 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java
@@ -71,10 +71,15 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
     /** Processed SQL statement text, may not be same as what user initially passed. */
     final String userSQL;
 
+    // flag whether is exec escape syntax
     private boolean isExecEscapeSyntax;
 
+    // flag whether is call escape syntax
     private boolean isCallEscapeSyntax;
 
+    // flag whether is four part syntax
+    private boolean isFourPartSyntax;
+
     /** Parameter positions in processed SQL statement text. */
     final int[] userSQLParamPositions;
 
@@ -144,6 +149,11 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
      */
     private static final Pattern execEscapePattern = Pattern.compile("^\\s*(?i)(?:exec|execute)\\b");
 
+    /**
+     * Regex for four part syntax
+     */
+    private static final Pattern fourPartSyntaxPattern = Pattern.compile("(.+)\\.(.+)\\.(.+)\\.(.+)");
+
     /** Returns the prepared statement SQL */
     @Override
     public String toString() {
@@ -271,6 +281,7 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
         userSQL = parsedSQL.processedSQL;
         isExecEscapeSyntax = isExecEscapeSyntax(sql);
         isCallEscapeSyntax = isCallEscapeSyntax(sql);
+        isFourPartSyntax = isFourPartSyntax(sql);
         userSQLParamPositions = parsedSQL.parameterPositions;
         initParams(userSQLParamPositions.length);
         useBulkCopyForBatchInsert = conn.getUseBulkCopyForBatchInsert();
@@ -1234,10 +1245,12 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
         // 2. There must be parameters
         // 3. Parameters must not be a TVP type
         // 4. Compliant CALL escape syntax
-        // If isExecEscapeSyntax is true, EXEC escape syntax is used then use prior behaviour to
-        // execute the procedure
+        // If isExecEscapeSyntax is true, EXEC escape syntax is used then use prior behaviour of
+        // wrapping call to execute the procedure
+        // If isFourPartSyntax is true, sproc is being executed against linked server, then
+        // use prior behaviour of wrapping call to execute procedure
         return (null != procedureName && paramCount != 0 && !isTVPType(params) && isCallEscapeSyntax
-                && !isExecEscapeSyntax);
+                && !isExecEscapeSyntax && !isFourPartSyntax);
     }
 
     /**
@@ -1265,6 +1278,10 @@ public class SQLServerPreparedStatement extends SQLServerStatement implements IS
         return callEscapePattern.matcher(sql).find();
     }
 
+    private boolean isFourPartSyntax(String sql) {
+        return fourPartSyntaxPattern.matcher(sql).find();
+    }
+
     /**
      * Executes sp_prepare to prepare a parameterized statement and sets the prepared statement handle
      *
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java
index 95b5ffa0..6cd502b7 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerResource.java
@@ -545,7 +545,8 @@ public final class SQLServerResource extends ListResourceBundle {
         {"R_ManagedIdentityTokenAcquisitionFail", "Failed to acquire managed identity token. Request for the token succeeded, but no token was returned. The token is null."},
         {"R_AmbiguousRowUpdate", "Failed to execute updateRow(). The update is attempting an ambiguous update on tables \"{0}\" and \"{1}\". Ensure all columns being updated prior to the updateRow() call belong to the same table."},
         {"R_InvalidSqlQuery", "Invalid SQL Query: {0}"},
-        {"R_InvalidScale", "Scale of input value is larger than the maximum allowed by SQL Server."}
+        {"R_InvalidScale", "Scale of input value is larger than the maximum allowed by SQL Server."},
+        {"R_colCountNotMatchColTypeCount", "Number of provided columns {0} does not match the column data types definition {1}."},
     };
 }
 // @formatter:on
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java
index 866b9ea1..cbfe102c 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java
@@ -1620,7 +1620,6 @@ public class SQLServerStatement implements ISQLServerStatement {
                         }
                     }
                 }
-
                 // If the current command (whatever it was) produced an error then stop parsing and propagate it up.
                 // In this case, the command is likely to be a RAISERROR, but it could be anything.
                 if (doneToken.isError())
@@ -1690,8 +1689,8 @@ public class SQLServerStatement implements ISQLServerStatement {
 
             @Override
             boolean onInfo(TDSReader tdsReader) throws SQLServerException {
-                StreamInfo infoToken = new StreamInfo();
-                infoToken.setFromTDS(tdsReader);
+                SQLServerInfoMessage infoMessage = new SQLServerInfoMessage();
+                infoMessage.setFromTDS(tdsReader);
 
                 // Under some circumstances the server cannot produce the cursored result set
                 // that we requested, but produces a client-side (default) result set instead.
@@ -1705,13 +1704,40 @@ public class SQLServerStatement implements ISQLServerStatement {
                 // ErrorCause: Server cursor is not supported on the specified SQL, falling back to default result set
                 // ErrorCorrectiveAction: None required
                 //
-                if (16954 == infoToken.msg.getErrorNumber())
+                if (16954 == infoMessage.msg.getErrorNumber())
                     executedSqlDirectly = true;
 
-                SQLWarning warning = new SQLWarning(
-                        infoToken.msg.getErrorMessage(), SQLServerException.generateStateCode(connection,
-                                infoToken.msg.getErrorNumber(), infoToken.msg.getErrorState()),
-                        infoToken.msg.getErrorNumber());
+
+                // Call the message handler to see what that think of the message
+                // - discard
+                // - upgrade to Error
+                // - or simply pass on
+                ISQLServerMessageHandler msgHandler = ((ISQLServerConnection)getConnection()).getServerMessageHandler();
+                if (msgHandler != null) {
+
+                    // Let the message handler decide if the error should be unchanged, up/down-graded or ignored
+                    ISQLServerMessage srvMessage = msgHandler.messageHandler(infoMessage);
+                
+                    // Ignored
+                    if (srvMessage == null) {
+                    	return true;
+                    }
+                
+                    // The message handler changed it to an "Error Message"
+                    if (srvMessage.isErrorMessage()) {
+                        // Set/Add the error message to the "super"
+                        addDatabaseError( (SQLServerError)srvMessage );
+                        return true;
+                    }
+                    
+                    // Still a "info message", just set infoMessage and the code in the below section will create the Warnings
+                    if (srvMessage.isInfoMessage()) {
+                        infoMessage = (SQLServerInfoMessage)srvMessage;
+                    }
+                }
+
+                // Create the SQLWarning and add them to the Warning chain
+                SQLWarning warning = new SQLServerWarning(infoMessage.msg);
 
                 if (sqlWarnings == null) {
                     sqlWarnings = new Vector<>();
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerWarning.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerWarning.java
new file mode 100644
index 00000000..81d588bd
--- /dev/null
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerWarning.java
@@ -0,0 +1,47 @@
+/*
+ * 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.sql.SQLWarning;
+
+/**
+ * Holds information about SQL Server messages that is considered as Informational Messages (normally if SQL Server Severity is at 10)
+ * <p>
+ * Instead of just holding the SQL Server message (like a normal SQLWarning, it also holds all the 
+ * SQL Servers extended information, like: ErrorSeverity, ServerName, ProcName etc
+ * <p>
+ * This enables client to print out extra information about the message.<br>
+ * Like: In what procedure was the message produced.
+ */
+public class SQLServerWarning
+extends SQLWarning
+{
+    private static final long serialVersionUID = -5212432397705929142L;
+
+    /** SQL server error */
+    private SQLServerError sqlServerError;
+    
+    /*
+     * Create a SQLWarning from an SQLServerError object
+     */
+    public SQLServerWarning(SQLServerError sqlServerError) {
+        super(sqlServerError.getErrorMessage(), 
+            SQLServerException.generateStateCode(null, sqlServerError.getErrorNumber(), sqlServerError.getErrorState()), 
+            sqlServerError.getErrorNumber(), 
+            null);
+
+        this.sqlServerError = sqlServerError;
+    }
+
+    /**
+     * Returns SQLServerError object containing detailed info about exception as received from SQL Server. This API
+     * returns null if no server error has occurred.
+     * 
+     * @return SQLServerError
+     */
+    public SQLServerError getSQLServerError() {
+        return sqlServerError;
+    }
+}
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerXAResource.java b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerXAResource.java
index 6d82bf83..eb96becb 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerXAResource.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerXAResource.java
@@ -5,7 +5,6 @@
 
 package com.microsoft.sqlserver.jdbc;
 
-import java.net.SocketException;
 import java.sql.CallableStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -25,7 +24,6 @@ import javax.transaction.xa.XAException;
 import javax.transaction.xa.XAResource;
 import javax.transaction.xa.Xid;
 
-import com.microsoft.sqlserver.jdbc.SQLServerError.TransientError;
 
 
 /**
@@ -754,27 +752,13 @@ public final class SQLServerXAResource implements javax.transaction.xa.XAResourc
                     }
                 }
             }
-        } catch (SQLTimeoutException ex) {
+        } catch (SQLTimeoutException | SQLServerException ex) {
             if (xaLogger.isLoggable(Level.FINER))
                 xaLogger.finer(toString() + " exception:" + ex);
             XAException e = new XAException(ex.toString());
             e.errorCode = XAException.XAER_RMFAIL;
             throw e;
 
-        } catch (SQLServerException ex) {
-            if (xaLogger.isLoggable(Level.FINER))
-                xaLogger.finer(toString() + " exception:" + ex);
-
-            if (ex.getMessage().equals(SQLServerException.getErrString("R_noServerResponse"))
-                    || TransientError.isTransientError(ex.getSQLServerError()) || isResourceManagerFailure(ex)) {
-                XAException e = new XAException(ex.toString());
-                e.errorCode = XAException.XAER_RMFAIL;
-                throw e;
-            }
-
-            XAException e = new XAException(ex.toString());
-            e.errorCode = XAException.XAER_RMERR;
-            throw e;
         }
 
         if (xaLogger.isLoggable(Level.FINER))
@@ -945,68 +929,4 @@ public final class SQLServerXAResource implements javax.transaction.xa.XAResourc
         return baseResourceID.incrementAndGet();
     }
 
-    private enum ResourceManagerFailure {
-        CONN_RESET("Connection reset"),
-        CONN_RESET_BY_PEER("Connection reset by peer"),
-        CONN_TIMEOUT("Connection timed out"),
-        CONN_RESILIENCY_CLIENT_UNRECOVERABLE(SQLServerException.getErrString("R_crClientUnrecoverable"));
-
-        private final String errString;
-
-        ResourceManagerFailure(String errString) {
-            this.errString = errString;
-        }
-
-        @Override
-        public String toString() {
-            return errString;
-        }
-
-        static ResourceManagerFailure fromString(String errString) {
-            for (ResourceManagerFailure resourceManagerFailure : ResourceManagerFailure.values()) {
-                if (errString.equalsIgnoreCase(resourceManagerFailure.toString())) {
-                    return resourceManagerFailure;
-                }
-            }
-            return null;
-        }
-    }
-
-    /**
-     * Check if the root exception of the throwable should be a XAER_RMFAIL exception
-     *
-     * @param throwable
-     *        The exception to check if the root cause should be a XAER_RMFAIL
-     *
-     * @return True if XAER_RMFAIL, otherwise false
-     */
-    private boolean isResourceManagerFailure(Throwable throwable) {
-        Throwable root = Util.getRootCause(throwable);
-
-        if (null == root) {
-            return false;
-        }
-
-        if (xaLogger.isLoggable(Level.FINE)) {
-            xaLogger.fine(toString() + " Resource manager failure root exception: " + root);
-        }
-
-        ResourceManagerFailure err = ResourceManagerFailure.fromString(root.getMessage());
-
-        if (null == err) {
-            return false;
-        }
-
-        // Add as needed here for future XAER_RMFAIL exceptions
-        switch (err) {
-            case CONN_RESET:
-            case CONN_RESET_BY_PEER:
-            case CONN_TIMEOUT:
-            case CONN_RESILIENCY_CLIENT_UNRECOVERABLE:
-                return true;
-            default:
-                return false;
-        }
-    }
-
 }
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/StreamInfo.java b/src/main/java/com/microsoft/sqlserver/jdbc/StreamInfo.java
deleted file mode 100644
index 5b4cb571..00000000
--- a/src/main/java/com/microsoft/sqlserver/jdbc/StreamInfo.java
+++ /dev/null
@@ -1,20 +0,0 @@
-/*
- * 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;
-
-final class StreamInfo extends StreamPacket {
-    final SQLServerError msg = new SQLServerError();
-
-    StreamInfo() {
-        super(TDS.TDS_MSG);
-    }
-
-    void setFromTDS(TDSReader tdsReader) throws SQLServerException {
-        if (TDS.TDS_MSG != tdsReader.readUnsignedByte())
-            assert false;
-        msg.setContentsFromTDS(tdsReader);
-    }
-}
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/Util.java b/src/main/java/com/microsoft/sqlserver/jdbc/Util.java
index 70b9c9aa..61bda22b 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/Util.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/Util.java
@@ -13,8 +13,6 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.text.DecimalFormat;
 import java.text.MessageFormat;
-import java.util.List;
-import java.util.ArrayList;
 import java.util.Date;
 import java.util.Locale;
 import java.util.Properties;
@@ -1046,23 +1044,6 @@ final class Util {
         }
         return chars;
     }
-
-    /**
-     * @param throwable
-     *        The exception to find root cause for.
-     *
-     * @return The root cause of the exception, otherwise null if null throwable input
-     */
-    static Throwable getRootCause(Throwable throwable) {
-        final List<Throwable> list = new ArrayList<>();
-
-        while (throwable != null && !list.contains(throwable)) {
-            list.add(throwable);
-            throwable = throwable.getCause();
-        }
-
-        return list.isEmpty() ? null : list.get(list.size() - 1);
-    }
 }
 
 
diff --git a/src/main/java/com/microsoft/sqlserver/jdbc/tdsparser.java b/src/main/java/com/microsoft/sqlserver/jdbc/tdsparser.java
index c18d61a0..90eebb8d 100644
--- a/src/main/java/com/microsoft/sqlserver/jdbc/tdsparser.java
+++ b/src/main/java/com/microsoft/sqlserver/jdbc/tdsparser.java
@@ -197,6 +197,14 @@ class TDSTokenHandler {
         return databaseError;
     }
 
+    public void addDatabaseError(SQLServerError databaseError) {
+        if (this.databaseError == null) {
+            this.databaseError = databaseError;
+        } else {
+            this.databaseError.addError(databaseError);
+        }
+    }
+
     TDSTokenHandler(String logContext) {
         this.logContext = logContext;
     }
@@ -258,13 +266,29 @@ class TDSTokenHandler {
     }
 
     boolean onError(TDSReader tdsReader) throws SQLServerException {
-        if (null == databaseError) {
-            databaseError = new SQLServerError();
-            databaseError.setFromTDS(tdsReader);
-        } else {
-            (new SQLServerError()).setFromTDS(tdsReader);
+        SQLServerError tmpDatabaseError = new SQLServerError();
+        tmpDatabaseError.setFromTDS(tdsReader);
+
+        ISQLServerMessageHandler msgHandler = tdsReader.getConnection().getServerMessageHandler();
+        if (msgHandler != null) {
+            // Let the message handler decide if the error should be unchanged/down-graded or ignored
+            ISQLServerMessage srvMessage = msgHandler.messageHandler(tmpDatabaseError);
+
+            // Ignored
+            if (srvMessage == null) {
+                return true;
+            }
+
+            // Down-graded to a SQLWarning
+            if (srvMessage.isInfoMessage()) {
+                tdsReader.getConnection().addWarning(srvMessage);
+                return true;
+            }
         }
 
+        // set/add the database error
+        addDatabaseError(tmpDatabaseError);
+
         return true;
     }
 
diff --git a/src/main/java/microsoft/sql/DateTimeOffset.java b/src/main/java/microsoft/sql/DateTimeOffset.java
index e2fcc840..bf9e95c7 100644
--- a/src/main/java/microsoft/sql/DateTimeOffset.java
+++ b/src/main/java/microsoft/sql/DateTimeOffset.java
@@ -72,6 +72,22 @@ public final class DateTimeOffset implements java.io.Serializable, java.lang.Com
         assert 0 == this.utcMillis % 1000L : "utcMillis: " + this.utcMillis;
     }
 
+    /**
+     * Constructs a DateTimeOffset from an existing java.time.OffsetDateTime
+     * DateTimeOffset represents values to 100 nanosecond precision. If the java.time.OffsetDateTime instance
+     * represents a value that is more precise, the value is rounded to the nearest multiple of 100 nanoseconds. Values
+     * within 50 nanoseconds of the next second are rounded up to the next second.
+     *
+     * @param offsetDateTime
+     *        A java.time.OffsetDateTime value
+     */
+    private DateTimeOffset(java.time.OffsetDateTime offsetDateTime) {
+        int hundredNanos = ((offsetDateTime.getNano() + 50) / 100);
+        this.utcMillis = (offsetDateTime.toEpochSecond() * 1000) + (hundredNanos / HUNDRED_NANOS_PER_SECOND * 1000);
+        this.nanos = 100 * (hundredNanos % HUNDRED_NANOS_PER_SECOND);
+        this.minutesOffset = offsetDateTime.getOffset().getTotalSeconds() / 60;
+    }
+
     /**
      * Converts a java.sql.Timestamp value with an integer offset to the equivalent DateTimeOffset value
      * 
@@ -105,6 +121,20 @@ public final class DateTimeOffset implements java.io.Serializable, java.lang.Com
                 (calendar.get(Calendar.ZONE_OFFSET) + calendar.get(Calendar.DST_OFFSET)) / (60 * 1000));
     }
 
+    /**
+     * Directly converts a {@link java.time.OffsetDateTime} value to an equivalent {@link DateTimeOffset} value
+     * DateTimeOffset represents values to 100 nanosecond precision. If the java.time.OffsetDateTime instance
+     * represents a value that is more precise, the value is rounded to the nearest multiple of 100 nanoseconds. Values
+     * within 50 nanoseconds of the next second are rounded up to the next second.
+     *
+     * @param offsetDateTime
+     *        A java.time.OffsetDateTime value
+     * @return The DateTimeOffset value of the input java.time.OffsetDateTime
+     */
+    public static DateTimeOffset valueOf(java.time.OffsetDateTime offsetDateTime) {
+        return new DateTimeOffset(offsetDateTime);
+    }
+
     /** formatted value */
     private String formattedValue = null;
 


Test patch:
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionTest.java
index eabbbb57..12eea46c 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/SQLServerConnectionTest.java
@@ -53,6 +53,7 @@ public class SQLServerConnectionTest extends AbstractTest {
     // If no retry is done, the function should at least exit in 5 seconds
     static int threshHoldForNoRetryInMilliseconds = 5000;
     static int loginTimeOutInSeconds = 10;
+    static String tnirHost = getConfiguredProperty("tnirHost");
 
     String randomServer = RandomUtil.getIdentifier("Server");
 
@@ -489,6 +490,44 @@ public class SQLServerConnectionTest extends AbstractTest {
         }
     }
 
+    // Test connect retry 0 but should still connect to TNIR
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    @Tag(Constants.xAzureSQLDB)
+    @Tag(Constants.reqExternalSetup)
+    public void testConnectTnir() {
+        org.junit.Assume.assumeTrue(isWindows);
+
+        // no retries but should connect to TNIR (this assumes host is defined in host file
+        try (Connection con = PrepUtil
+                .getConnection(connectionString + ";transparentNetworkIPResolution=true;connectRetryCount=0;serverName="
+                        + tnirHost);) {} catch (Exception e) {
+            fail(e.getMessage());
+        }
+    }
+
+    // Test connect retry 0 and TNIR disabled
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    @Tag(Constants.xAzureSQLDB)
+    @Tag(Constants.reqExternalSetup)
+    public void testConnectNoTnir() {
+        org.junit.Assume.assumeTrue(isWindows);
+
+        // no retries no TNIR should fail even tho host is defined in host file
+        try (Connection con = PrepUtil.getConnection(connectionString
+                + ";transparentNetworkIPResolution=false;connectRetryCount=0;serverName=" + tnirHost);) {
+            assertTrue(con == null, TestResource.getResource("R_shouldNotConnect"));
+        } catch (Exception e) {
+            assertTrue(e.getMessage().matches(TestUtils.formatErrorMsg("R_tcpipConnectionFailed"))
+                    || ((isSqlAzure() || isSqlAzureDW())
+                                                         ? e.getMessage().contains(
+                                                                 TestResource.getResource("R_connectTimedOut"))
+                                                         : false),
+                    e.getMessage());
+        }
+    }
+
     @Test
     @Tag(Constants.xAzureSQLDW)
     @Tag(Constants.xAzureSQLDB)
@@ -981,7 +1020,9 @@ public class SQLServerConnectionTest extends AbstractTest {
 
                 ds.setURL(connectionString);
                 ds.setServerName("invalidServerName" + UUID.randomUUID());
-                ds.setLoginTimeout(5);
+                ds.setLoginTimeout(30);
+                ds.setConnectRetryCount(3);
+                ds.setConnectRetryInterval(10);
                 try (Connection con = ds.getConnection()) {} catch (SQLException e) {}
             }
         };
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java b/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java
index e14bb671..aa4252f9 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/TestResource.java
@@ -63,6 +63,11 @@ public final class TestResource extends ListResourceBundle {
             {"R_ConnectionURLNull", "The connection URL is null."},
             {"R_connectionIsNotClosed", "The connection is not closed."},
             {"R_invalidExceptionMessage", "Invalid exception message"},
+            {"R_invalidClientSecret", "AADSTS7000215: Invalid client secret provided"},
+            {"R_invalidCertFields",
+                    "Error reading certificate, please verify the location of the certificate.signed fields invalid"},
+            {"R_invalidAADAuth",
+                    "Failed to authenticate the user {0} in Active Directory (Authentication={1})"},
             {"R_failedValidate", "failed to validate values in $0} "}, {"R_tableNotDropped", "table not dropped. "},
             {"R_connectionReset", "Connection reset"}, {"R_unknownException", "Unknown exception"},
             {"R_deadConnection", "Dead connection should be invalid"},
@@ -212,5 +217,7 @@ public final class TestResource extends ListResourceBundle {
             {"R_failedFedauth", "Failed to acquire fedauth token: "},
             {"R_noLoginModulesConfiguredForJdbcDriver",
                     "javax.security.auth.login.LoginException (No LoginModules configured for SQLJDBCDriver)"},
-            {"R_unexpectedThreadCount", "Thread count is higher than expected."}};
+            {"R_unexpectedThreadCount", "Thread count is higher than expected."},
+            {"R_expectedClassDoesNotMatchActualClass",
+                    "Expected column class {0} does not match actual column class {1} for column {2}."}};
 }
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyCSVTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyCSVTest.java
index 48e28b71..97b0e3f9 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyCSVTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/bulkCopy/BulkCopyCSVTest.java
@@ -153,7 +153,7 @@ public class BulkCopyCSVTest extends AbstractTest {
         /*
          * The list below is the copy of inputFileDelimiterEsc ape with quotes removed.
          */
-        String[][] expectedEscaped = new String[11][4];
+        String[][] expectedEscaped = new String[12][4];
         expectedEscaped[0] = new String[] {"test", " test\"", "no@split", " testNoQuote", ""};
         expectedEscaped[1] = new String[] {null, null, null, null, ""};
         expectedEscaped[2] = new String[] {"\"", "test\"test", "test@\"  test", null, ""};
@@ -166,6 +166,7 @@ public class BulkCopyCSVTest extends AbstractTest {
         expectedEscaped[8] = new String[] {"1997", "Ford", "E350", "Super@ \"luxurious\" truck", ""};
         expectedEscaped[9] = new String[] {"1997", "Ford", "E350", "E63", ""};
         expectedEscaped[10] = new String[] {"1997", "Ford", "E350", " Super luxurious truck ", ""};
+        expectedEscaped[11] = new String[] {"1997", "F\r\no\r\nr\r\nd", "E350", "\"Super\" \"luxurious\" \"truck\"", ""};
 
         try (Connection con = getConnection(); Statement stmt = con.createStatement();
                 SQLServerBulkCopy bulkCopy = new SQLServerBulkCopy(con);
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/callablestatement/CallableStatementTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/callablestatement/CallableStatementTest.java
index 6570abf7..51203939 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/callablestatement/CallableStatementTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/callablestatement/CallableStatementTest.java
@@ -1150,6 +1150,66 @@ public class CallableStatementTest extends AbstractTest {
         }
     }
 
+    @Test
+    @Tag(Constants.reqExternalSetup)
+    @Tag(Constants.xAzureSQLDB)
+    @Tag(Constants.xAzureSQLDW)
+    @Tag(Constants.xAzureSQLMI)
+    public void testFourPartSyntaxCallEscapeSyntax() throws SQLException {
+        String table = "serverList";
+
+        try (Statement stmt = connection.createStatement()) {
+            stmt.execute("IF OBJECT_ID(N'" + table + "') IS NOT NULL DROP TABLE " + table);
+            stmt.execute("CREATE TABLE " + table + " (serverName varchar(100),network varchar(100),serverStatus varchar(4000), id int, collation varchar(100), connectTimeout int, queryTimeout int)");
+            stmt.execute("INSERT " + table + " EXEC sp_helpserver");
+
+            ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM " + table + " WHERE serverName = N'" + linkedServer + "'");
+            rs.next();
+
+            if (rs.getInt(1) == 1) {
+                stmt.execute("EXEC sp_dropserver @server='" + linkedServer + "';");
+            }
+
+            stmt.execute("EXEC sp_addlinkedserver @server='" + linkedServer + "';");
+            stmt.execute("EXEC sp_addlinkedsrvlogin @rmtsrvname=N'" + linkedServer + "', @rmtuser=N'" + remoteUser + "', @rmtpassword=N'" + remotePassword + "'");
+            stmt.execute("EXEC sp_serveroption '" + linkedServer + "', 'rpc', true;");
+            stmt.execute("EXEC sp_serveroption '" + linkedServer + "', 'rpc out', true;");
+        }
+
+        SQLServerDataSource ds = new SQLServerDataSource();
+        ds.setServerName(linkedServer);
+        ds.setUser(remoteUser);
+        ds.setPassword(remotePassword);
+        ds.setEncrypt(false);
+        ds.setTrustServerCertificate(true);
+
+        try (Connection linkedServerConnection = ds.getConnection(); Statement stmt = linkedServerConnection.createStatement()) {
+            stmt.execute("create or alter procedure dbo.TestAdd(@Num1 int, @Num2 int, @Result int output) as begin set @Result = @Num1 + @Num2; end;");
+        }
+
+        try (CallableStatement cstmt = connection.prepareCall("{call [" + linkedServer + "].master.dbo.TestAdd(?,?,?)}")) {
+            int sum = 11;
+            int param0 = 1;
+            int param1 = 10;
+            cstmt.setInt(1, param0);
+            cstmt.setInt(2, param1);
+            cstmt.registerOutParameter(3, Types.INTEGER);
+            cstmt.execute();
+            assertEquals(sum, cstmt.getInt(3));
+        }
+
+        try (CallableStatement cstmt = connection.prepareCall("exec [" + linkedServer + "].master.dbo.TestAdd ?,?,?")) {
+            int sum = 11;
+            int param0 = 1;
+            int param1 = 10;
+            cstmt.setInt(1, param0);
+            cstmt.setInt(2, param1);
+            cstmt.registerOutParameter(3, Types.INTEGER);
+            cstmt.execute();
+            assertEquals(sum, cstmt.getInt(3));
+        }
+    }
+
     /**
      * Cleanup after test
      *
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 d758950f..926b6a76 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/connection/RequestBoundaryMethodsTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/connection/RequestBoundaryMethodsTest.java
@@ -521,6 +521,10 @@ public class RequestBoundaryMethodsTest extends AbstractTest {
         verifiedMethodNames.add("setMsiTokenCacheTtl");
         verifiedMethodNames.add("getAccessTokenCallbackClass");
         verifiedMethodNames.add("setAccessTokenCallbackClass");
+        verifiedMethodNames.add("getServerMessageHandler");
+        verifiedMethodNames.add("setServerMessageHandler");
+        verifiedMethodNames.add("getCalcBigDecimalScale");
+        verifiedMethodNames.add("setCalcBigDecimalScale");
         verifiedMethodNames.add("getUseFlexibleCallableStatements");
         verifiedMethodNames.add("setUseFlexibleCallableStatements");
         verifiedMethodNames.add("getCalcBigDecimalPrecision");
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/connection/TimeoutTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/connection/TimeoutTest.java
index af2bda8a..268fda73 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/connection/TimeoutTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/connection/TimeoutTest.java
@@ -9,7 +9,6 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
 import java.lang.reflect.Field;
-import java.lang.reflect.Method;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
@@ -36,67 +35,81 @@ import com.microsoft.sqlserver.testframework.PrepUtil;
 
 
 @RunWith(JUnitPlatform.class)
-@Tag("slow")
 public class TimeoutTest extends AbstractTest {
     static String randomServer = RandomUtil.getIdentifier("Server");
     static String waitForDelaySPName = RandomUtil.getIdentifier("waitForDelaySP");
     static final int waitForDelaySeconds = 10;
-    static final int defaultTimeout = 15; // loginTimeout default value
+    static final int defaultTimeout = 30; // loginTimeout default value
 
     @BeforeAll
     public static void setupTests() throws Exception {
         setConnection();
     }
 
+    /*
+     * TODO:
+     * The tests below uses a simple interval counting logic to determine whether there was at least 1 retry.
+     * Given the interval is long enough, then 1 retry should take at least 1 interval long, so if it took < 1 interval, then it assumes there were no retry. However, this only works if TNIR or failover is not enabled since those cases should retry but no wait interval in between. So this interval counting can not detect these cases.
+     * Note a better and more reliable way would be to check attemptNumber using reflection to determine the number of retries.
+     */
+
+    // test default loginTimeout used if not specified in connection string
     @Test
     public void testDefaultLoginTimeout() {
-        long timerEnd = 0;
-
+        long totalTime = 0;
         long timerStart = System.currentTimeMillis();
-        // Try a non existing server and see if the default timeout is 15 seconds
-        try (Connection con = PrepUtil.getConnection("jdbc:sqlserver://" + randomServer + "connectRetryCount=0")) {
+
+        // non existing server and default values to see if took default timeout
+        try (Connection con = PrepUtil.getConnection("jdbc:sqlserver://" + randomServer)) {
             fail(TestResource.getResource("R_shouldNotConnect"));
         } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
+            totalTime = System.currentTimeMillis() - timerStart;
 
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_tcpipConnectionToHost")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
+            assertTrue(
+                    (e.getMessage().toLowerCase()
+                            .contains(TestResource.getResource("R_tcpipConnectionToHost").toLowerCase()))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                     e.getMessage());
         }
 
-        verifyTimeout(timerEnd - timerStart, defaultTimeout);
+        // time should be < default loginTimeout
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(defaultTimeout),
+                "total time: " + totalTime + " default loginTimout: " + TimeUnit.SECONDS.toMillis(defaultTimeout));
     }
 
+    // test setting loginTimeout value
     @Test
     public void testURLLoginTimeout() {
-        long timerEnd = 0;
-        int timeout = 10;
+        long totalTime = 0;
+        int timeout = 15;
 
         long timerStart = System.currentTimeMillis();
 
+        // non existing server and set loginTimeout
         try (Connection con = PrepUtil.getConnection("jdbc:sqlserver://" + randomServer + ";logintimeout=" + timeout)) {
             fail(TestResource.getResource("R_shouldNotConnect"));
         } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
+            totalTime = System.currentTimeMillis() - timerStart;
 
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_tcpipConnectionToHost")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
+            assertTrue(
+                    (e.getMessage().toLowerCase()
+                            .contains(TestResource.getResource("R_tcpipConnectionToHost").toLowerCase()))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                     e.getMessage());
         }
 
-        verifyTimeout(timerEnd - timerStart, timeout);
+        // time should be < set loginTimeout
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(timeout),
+                "total time: " + totalTime + " loginTimeout: " + TimeUnit.SECONDS.toMillis(timeout));
     }
 
+    // test setting timeout in DM
     @Test
     public void testDMLoginTimeoutApplied() {
-        long timerEnd = 0;
-        int timeout = 10;
+        long totalTime = 0;
+        int timeout = 15;
 
         DriverManager.setLoginTimeout(timeout);
         long timerStart = System.currentTimeMillis();
@@ -104,23 +117,26 @@ public class TimeoutTest extends AbstractTest {
         try (Connection con = PrepUtil.getConnection("jdbc:sqlserver://" + randomServer)) {
             fail(TestResource.getResource("R_shouldNotConnect"));
         } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
+            totalTime = System.currentTimeMillis() - timerStart;
 
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_tcpipConnectionToHost")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
+            assertTrue(
+                    (e.getMessage().toLowerCase()
+                            .contains(TestResource.getResource("R_tcpipConnectionToHost").toLowerCase()))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                     e.getMessage());
         }
 
-        verifyTimeout(timerEnd - timerStart, timeout);
+        // time should be < DM timeout
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(timeout),
+                "total time: " + totalTime + " DM loginTimeout: " + TimeUnit.SECONDS.toMillis(timeout));
     }
 
+    // test that setting in connection string overrides value set in DM
     @Test
     public void testDMLoginTimeoutNotApplied() {
-        long timerEnd = 0;
-        int timeout = 10;
+        long totalTime = 0;
+        int timeout = 15;
         try {
             DriverManager.setLoginTimeout(timeout * 3); // 30 seconds
             long timerStart = System.currentTimeMillis();
@@ -129,134 +145,185 @@ public class TimeoutTest extends AbstractTest {
                     .getConnection("jdbc:sqlserver://" + randomServer + ";loginTimeout=" + timeout)) {
                 fail(TestResource.getResource("R_shouldNotConnect"));
             } catch (Exception e) {
-                timerEnd = System.currentTimeMillis();
+                totalTime = System.currentTimeMillis() - timerStart;
 
                 assertTrue(
-                        (e.getMessage().contains(TestResource.getResource("R_tcpipConnectionToHost")))
-                                || ((isSqlAzure() || isSqlAzureDW())
-                                                                     ? e.getMessage()
-                                                                             .contains(TestResource
-                                                                                     .getResource("R_connectTimedOut"))
-                                                                     : false),
+                        (e.getMessage().toLowerCase()
+                                .contains(TestResource.getResource("R_tcpipConnectionToHost").toLowerCase()))
+                                || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                        .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                         e.getMessage());
             }
-            verifyTimeout(timerEnd - timerStart, timeout);
+
+            // time should be < connection string loginTimeout
+            assertTrue(totalTime < TimeUnit.SECONDS.toMillis(timeout),
+                    "total time: " + totalTime + " loginTimeout: " + TimeUnit.SECONDS.toMillis(timeout));
         } finally {
             DriverManager.setLoginTimeout(0); // Default to 0 again
         }
     }
 
+    // Test connect retry set to 0 (disabled)
+    @Test
+    public void testConnectRetryDisable() {
+        long totalTime = 0;
+        long timerStart = System.currentTimeMillis();
+        int interval = defaultTimeout; // long interval so we can tell if there was a retry
+        long timeout = defaultTimeout * 2; // long loginTimeout to accommodate the long interval
+
+        // non existent server with long loginTimeout, should return fast if no retries at all
+        try (Connection con = PrepUtil.getConnection(
+                "jdbc:sqlserver://" + randomServer + ";transparentNetworkIPResolution=false;loginTimeout=" + timeout
+                        + ";connectRetryCount=0;connectInterval=" + interval)) {
+            fail(TestResource.getResource("R_shouldNotConnect"));
+        } catch (Exception e) {
+            totalTime = System.currentTimeMillis() - timerStart;
+
+            assertTrue(
+                    e.getMessage().matches(TestUtils.formatErrorMsg("R_tcpipConnectionFailed"))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
+                    e.getMessage());
+        }
+
+        // if there was a retry then it would take at least 1 interval long, so if < interval means there were no retries
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(interval),
+                "total time: " + totalTime + " interval: " + TimeUnit.SECONDS.toMillis(interval));
+    }
+
     // Test connect retry for non-existent server with loginTimeout
     @Test
     public void testConnectRetryBadServer() {
-        long timerEnd = 0;
+        long totalTime = 0;
         long timerStart = System.currentTimeMillis();
-        int loginTimeout = 15;
+        int timeout = 15;
 
         // non existent server with very short loginTimeout, no retry will happen as not a transient error
-        try (Connection con = PrepUtil
-                .getConnection("jdbc:sqlserver://" + randomServer + ";loginTimeout=" + loginTimeout)) {
+        try (Connection con = PrepUtil.getConnection("jdbc:sqlserver://" + randomServer + ";loginTimeout=" + timeout)) {
             fail(TestResource.getResource("R_shouldNotConnect"));
         } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
+            totalTime = System.currentTimeMillis() - timerStart;
 
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_tcpipConnectionToHost")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
+            assertTrue(
+                    (e.getMessage().toLowerCase()
+                            .contains(TestResource.getResource("R_tcpipConnectionToHost").toLowerCase()))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                     e.getMessage());
         }
 
-        verifyTimeout(timerEnd - timerStart, loginTimeout);
+        // time should be < loginTimeout set
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(timeout),
+                "total time: " + totalTime + " loginTimeout: " + TimeUnit.SECONDS.toMillis(timeout));
     }
 
     // Test connect retry for database error
     @Test
     public void testConnectRetryServerError() {
-        long timerEnd = 0;
+        long totalTime = 0;
         long timerStart = System.currentTimeMillis();
+        int interval = defaultTimeout; // long interval so we can tell if there was a retry
+        long timeout = defaultTimeout * 2; // long loginTimeout to accommodate the long interval
 
-        // non existent database with interval < loginTimeout this will generate a 4060 transient error and retry
-        int connectRetryCount = new Random().nextInt(256);
-        int connectRetryInterval = new Random().nextInt(defaultTimeout) + 1;
+        // non existent database with interval < loginTimeout this will generate a 4060 transient error and retry 1 time
         try (Connection con = PrepUtil.getConnection(
                 TestUtils.addOrOverrideProperty(connectionString, "database", RandomUtil.getIdentifier("database"))
-                        + ";logintimeout=" + defaultTimeout + ";connectRetryCount=" + connectRetryCount
-                        + ";connectRetryInterval=" + connectRetryInterval)) {
+                        + ";loginTimeout=" + timeout + ";connectRetryCount=" + 1 + ";connectRetryInterval=" + interval
+                        + ";transparentNetworkIPResolution=false")) {
             fail(TestResource.getResource("R_shouldNotConnect"));
         } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
+            totalTime = System.currentTimeMillis() - timerStart;
 
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_cannotOpenDatabase")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
+            assertTrue(
+                    (e.getMessage().toLowerCase()
+                            .contains(TestResource.getResource("R_cannotOpenDatabase").toLowerCase()))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                     e.getMessage());
         }
 
-        // connect + all retries should always be <= loginTimeout
-        verifyTimeout(timerEnd - timerStart, defaultTimeout);
+        // 1 retry should be at least 1 interval long but < 2 intervals
+        assertTrue(TimeUnit.SECONDS.toMillis(interval) < totalTime,
+                "interval: " + TimeUnit.SECONDS.toMillis(interval) + " total time: " + totalTime);
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(2 * interval),
+                "total time: " + totalTime + " 2 * interval: " + TimeUnit.SECONDS.toMillis(interval));
     }
 
     // Test connect retry for database error using Datasource
     @Test
     public void testConnectRetryServerErrorDS() {
-        long timerEnd = 0;
+        long totalTime = 0;
         long timerStart = System.currentTimeMillis();
+        int interval = defaultTimeout; // long interval so we can tell if there was a retry
+        long loginTimeout = defaultTimeout * 2; // long loginTimeout to accommodate the long interval
 
-        // non existent database with interval < loginTimeout this will generate a 4060 transient error and retry
-        int connectRetryCount = new Random().nextInt(256);
-        int connectRetryInterval = new Random().nextInt(defaultTimeout) + 1;
-
+        // non existent database with interval < loginTimeout this will generate a 4060 transient error and retry 1 time
         SQLServerDataSource ds = new SQLServerDataSource();
         String connectStr = TestUtils.addOrOverrideProperty(connectionString, "database",
-                RandomUtil.getIdentifier("database")) + ";logintimeout=" + defaultTimeout + ";connectRetryCount="
-                + connectRetryCount + ";connectRetryInterval=" + connectRetryInterval;
+                RandomUtil.getIdentifier("database")) + ";logintimeout=" + loginTimeout + ";connectRetryCount=1"
+                + ";connectRetryInterval=" + interval;
         updateDataSource(connectStr, ds);
 
         try (Connection con = PrepUtil.getConnection(connectStr)) {
             fail(TestResource.getResource("R_shouldNotConnect"));
         } catch (Exception e) {
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_cannotOpenDatabase")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
+            assertTrue(
+                    (e.getMessage().toLowerCase()
+                            .contains(TestResource.getResource("R_cannotOpenDatabase").toLowerCase()))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                     e.getMessage());
-            timerEnd = System.currentTimeMillis();
+            totalTime = System.currentTimeMillis() - timerStart;
         }
 
-        // connect + all retries should always be <= loginTimeout
-        verifyTimeout(timerEnd - timerStart, defaultTimeout);
+        // 1 retry should be at least 1 interval long but < 2 intervals
+        assertTrue(TimeUnit.SECONDS.toMillis(interval) < totalTime,
+                "interval: " + TimeUnit.SECONDS.toMillis(interval) + " total time: " + totalTime);
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(2 * interval),
+                "total time: " + totalTime + " 2 * interval: " + TimeUnit.SECONDS.toMillis(2 * interval));
     }
 
     // Test connect retry for database error with loginTimeout
     @Test
     public void testConnectRetryTimeout() {
-        long timerEnd = 0;
+        long totalTime = 0;
         long timerStart = System.currentTimeMillis();
+        int interval = defaultTimeout; // long interval so we can tell if there was a retry
         int loginTimeout = 2;
 
-        // non existent database with very short loginTimeout so there is no time to do all retries
+        // non existent database with very short loginTimeout so there is no time to do any retry
         try (Connection con = PrepUtil.getConnection(
                 TestUtils.addOrOverrideProperty(connectionString, "database", RandomUtil.getIdentifier("database"))
-                        + "connectRetryCount=" + (new Random().nextInt(256)) + ";connectRetryInterval="
-                        + (new Random().nextInt(defaultTimeout - 1) + 1) + ";loginTimeout=" + loginTimeout)) {
+                        + "connectRetryCount=" + (new Random().nextInt(256)) + ";connectRetryInterval=" + interval
+                        + ";loginTimeout=" + loginTimeout)) {
             fail(TestResource.getResource("R_shouldNotConnect"));
         } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
+            totalTime = System.currentTimeMillis() - timerStart;
 
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_cannotOpenDatabase")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
+            assertTrue(
+                    (e.getMessage().toLowerCase()
+                            .contains(TestResource.getResource("R_cannotOpenDatabase").toLowerCase()))
+                            || ((isSqlAzure() || isSqlAzureDW()) ? e.getMessage().toLowerCase()
+                                    .contains(TestResource.getResource("R_connectTimedOut").toLowerCase()) : false),
                     e.getMessage());
         }
 
-        verifyTimeout(timerEnd - timerStart, loginTimeout);
+        // if there was a retry then it would take at least 1 interval long, so if < interval means there were no retries
+        assertTrue(totalTime < TimeUnit.SECONDS.toMillis(interval),
+                "total time: " + totalTime + " interval: " + TimeUnit.SECONDS.toMillis(interval));
+    }
+
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    public void testSocketTimeoutBoundedByLoginTimeoutReset() throws Exception {
+        try (Connection con = PrepUtil.getConnection(connectionString + ";socketTimeout=90000;loginTimeout=10;");
+                Statement stmt = con.createStatement()) {
+            // Login timeout (10s) is less than the 15s sec WAITFOR DELAY. Upon a login attempt, socketTimeout should be bounded
+            // by loginTimeout. After a successful login, when executing a query, socketTimeout should be reset to the
+            // original 90000ms timeout. The statement below should successfully execute as socketTimeout should not be bounded
+            // by loginTimeout, otherwise the test fails with a socket read timeout error.
+            stmt.execute("WAITFOR DELAY '00:00:15';");
+        }
     }
 
     // Test for detecting Azure server for connection retries
@@ -289,80 +356,6 @@ public class TimeoutTest extends AbstractTest {
         }
     }
 
-    @Test
-    public void testFailoverInstanceResolution() throws SQLException {
-        long timerEnd = 0;
-        long timerStart = System.currentTimeMillis();
-
-        // Try a non existing server and see if the default timeout is 15 seconds
-        try (Connection con = PrepUtil
-                .getConnection("jdbc:sqlserver://" + randomServer + ";databaseName=FailoverDB_abc;failoverPartner="
-                        + randomServer + "\\foo;user=sa;password=" + RandomUtil.getIdentifier("password"))) {
-            fail(TestResource.getResource("R_shouldNotConnect"));
-        } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
-
-            assertTrue((e.getMessage().contains(TestResource.getResource("R_tcpipConnectionToHost")))
-                    || ((isSqlAzure() || isSqlAzureDW())
-                                                         ? e.getMessage().contains(
-                                                                 TestResource.getResource("R_connectTimedOut"))
-                                                         : false),
-                    e.getMessage());
-        }
-
-        verifyTimeout(timerEnd - timerStart, defaultTimeout * 2);
-    }
-
-    @Test
-    public void testFOInstanceResolution2() throws SQLException {
-        long timerEnd = 0;
-
-        long timerStart = System.currentTimeMillis();
-        try (Connection con = PrepUtil
-                .getConnection("jdbc:sqlserver://" + randomServer + "\\fooggg;databaseName=FailoverDB;failoverPartner="
-                        + randomServer + "\\foo;user=sa;password=" + RandomUtil.getIdentifier("password"))) {
-            fail(TestResource.getResource("R_shouldNotConnect"));
-        } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
-        }
-
-        verifyTimeout(timerEnd - timerStart, defaultTimeout);
-    }
-
-    /**
-     * Tests that failover is correctly used after a socket timeout, by confirming total time includes socketTimeout
-     * for both primary and failover server.
-     */
-    @Test
-    public void testFailoverInstanceResolutionWithSocketTimeout() {
-        long timerEnd;
-        long timerStart = System.currentTimeMillis();
-
-        try (Connection con = PrepUtil.getConnection("jdbc:sqlserver://" + randomServer
-                + ";databaseName=FailoverDB;failoverPartner=" + randomServer + "\\foo;user=sa;password="
-                + RandomUtil.getIdentifier("password") + ";socketTimeout=" + waitForDelaySeconds)) {
-            fail(TestResource.getResource("R_shouldNotConnect"));
-        } catch (Exception e) {
-            timerEnd = System.currentTimeMillis();
-            if (!(e instanceof SQLException)) {
-                fail(TestResource.getResource("R_unexpectedErrorMessage") + e.getMessage());
-            }
-
-            // Driver should correctly attempt to connect to db, experience a socketTimeout, attempt to connect to
-            // failover, and then have another socketTimeout. So, expected total time is 2 x socketTimeout.
-            long totalTime = timerEnd - timerStart;
-            long totalExpectedTime = waitForDelaySeconds * 1000L * 2; // We expect 2 * socketTimeout
-            assertTrue(totalTime >= totalExpectedTime, TestResource.getResource("R_executionNotLong") + "totalTime: "
-                    + totalTime + " expectedTime: " + totalExpectedTime);
-        }
-    }
-
-    private void verifyTimeout(long timeDiff, int timeout) {
-        // Verify that login timeout does not take longer than <timeout * 2> seconds.
-        assertTrue(timeDiff < TimeUnit.SECONDS.toMillis(timeout * 2),
-                "timeout: " + TimeUnit.SECONDS.toMillis(timeout) + " timediff: " + timeDiff);
-    }
-
     /**
      * When query timeout occurs, the connection is still usable.
      * 
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/databasemetadata/DatabaseMetaDataTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/databasemetadata/DatabaseMetaDataTest.java
index ff2f2fc7..999ccdc1 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/databasemetadata/DatabaseMetaDataTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/databasemetadata/DatabaseMetaDataTest.java
@@ -63,6 +63,38 @@ public class DatabaseMetaDataTest extends AbstractTest {
     private static final String functionName = RandomUtil.getIdentifier("DBMetadataFunction");
     private static Map<Integer, String> getColumnsDWColumns = null;
     private static Map<Integer, String> getImportedKeysDWColumns = null;
+    private static final String TABLE_CAT = "TABLE_CAT";
+    private static final String TABLE_SCHEM = "TABLE_SCHEM";
+    private static final String TABLE_NAME = "TABLE_NAME";
+    private static final String COLUMN_NAME = "COLUMN_NAME";
+    private static final String DATA_TYPE = "DATA_TYPE";
+    private static final String TYPE_NAME = "TYPE_NAME";
+    private static final String COLUMN_SIZE = "COLUMN_SIZE";
+    private static final String BUFFER_LENGTH = "BUFFER_LENGTH";
+    private static final String DECIMAL_DIGITS = "DECIMAL_DIGITS";
+    private static final String NUM_PREC_RADIX = "NUM_PREC_RADIX";
+    private static final String NULLABLE = "NULLABLE";
+    private static final String REMARKS = "REMARKS";
+    private static final String COLUMN_DEF = "COLUMN_DEF";
+    private static final String SQL_DATA_TYPE = "SQL_DATA_TYPE";
+    private static final String SQL_DATETIME_SUB = "SQL_DATETIME_SUB";
+    private static final String CHAR_OCTET_LENGTH = "CHAR_OCTET_LENGTH";
+    private static final String ORDINAL_POSITION = "ORDINAL_POSITION";
+    private static final String IS_NULLABLE = "IS_NULLABLE";
+    private static final String SCOPE_CATALOG = "SCOPE_CATALOG";
+    private static final String SCOPE_SCHEMA = "SCOPE_SCHEMA";
+    private static final String SCOPE_TABLE = "SCOPE_TABLE";
+    private static final String SOURCE_DATA_TYPE = "SOURCE_DATA_TYPE";
+    private static final String IS_AUTOINCREMENT = "IS_AUTOINCREMENT";
+    private static final String IS_GENERATEDCOLUMN = "IS_GENERATEDCOLUMN";
+    private static final String SS_IS_SPARSE = "SS_IS_SPARSE";
+    private static final String SS_IS_COLUMN_SET = "SS_IS_COLUMN_SET";
+    private static final String SS_UDT_CATALOG_NAME = "SS_UDT_CATALOG_NAME";
+    private static final String SS_UDT_SCHEMA_NAME = "SS_UDT_SCHEMA_NAME";
+    private static final String SS_UDT_ASSEMBLY_TYPE_NAME = "SS_UDT_ASSEMBLY_TYPE_NAME";
+    private static final String SS_XML_SCHEMACOLLECTION_CATALOG_NAME = "SS_XML_SCHEMACOLLECTION_CATALOG_NAME";
+    private static final String SS_XML_SCHEMACOLLECTION_SCHEMA_NAME = "SS_XML_SCHEMACOLLECTION_SCHEMA_NAME";
+    private static final String SS_XML_SCHEMACOLLECTION_NAME = "SS_XML_SCHEMACOLLECTION_NAME";
 
     /**
      * Verify DatabaseMetaData#isWrapperFor and DatabaseMetaData#unwrap.
@@ -887,6 +919,78 @@ public class DatabaseMetaDataTest extends AbstractTest {
         }
     }
 
+    /**
+     * Validates the metadata data types defined by JDBC spec.
+     * Refer to <a href="https://docs.oracle.com/javase/8/docs/api/java/sql/DatabaseMetaData.html#getColumns-java.lang.String-java.lang.String-java.lang.String-java.lang.String-">DatabaseMetadata getColumns() specs</a>
+     * 
+     * @throws SQLException
+     */
+    @Test
+    public void testValidateColumnMetadata() throws SQLException {
+        Map<String, Class<?>> getColumnMetaDataClass = new LinkedHashMap<>();
+
+        getColumnMetaDataClass.put(TABLE_CAT, String.class);
+        getColumnMetaDataClass.put(TABLE_SCHEM, String.class);
+        getColumnMetaDataClass.put(TABLE_NAME, String.class);
+        getColumnMetaDataClass.put(COLUMN_NAME, String.class);
+        getColumnMetaDataClass.put(DATA_TYPE, Integer.class);
+        getColumnMetaDataClass.put(TYPE_NAME, String.class);
+        getColumnMetaDataClass.put(COLUMN_SIZE, Integer.class);
+        getColumnMetaDataClass.put(BUFFER_LENGTH, Integer.class); // Not used
+        getColumnMetaDataClass.put(DECIMAL_DIGITS, Integer.class);
+        getColumnMetaDataClass.put(NUM_PREC_RADIX, Integer.class);
+        getColumnMetaDataClass.put(NULLABLE, Integer.class);
+        getColumnMetaDataClass.put(REMARKS, String.class);
+        getColumnMetaDataClass.put(COLUMN_DEF, String.class);
+        getColumnMetaDataClass.put(SQL_DATA_TYPE, Integer.class);
+        getColumnMetaDataClass.put(SQL_DATETIME_SUB, Integer.class);
+        getColumnMetaDataClass.put(CHAR_OCTET_LENGTH, Integer.class);
+        getColumnMetaDataClass.put(ORDINAL_POSITION, Integer.class);
+        getColumnMetaDataClass.put(IS_NULLABLE, String.class);
+        getColumnMetaDataClass.put(SCOPE_CATALOG, String.class);
+        getColumnMetaDataClass.put(SCOPE_SCHEMA, String.class);
+        getColumnMetaDataClass.put(SCOPE_TABLE, String.class);
+        getColumnMetaDataClass.put(SOURCE_DATA_TYPE, Short.class);
+        getColumnMetaDataClass.put(IS_AUTOINCREMENT, String.class);
+        getColumnMetaDataClass.put(IS_GENERATEDCOLUMN, String.class);
+        getColumnMetaDataClass.put(SS_IS_SPARSE, Short.class);
+        getColumnMetaDataClass.put(SS_IS_COLUMN_SET, Short.class);
+        getColumnMetaDataClass.put(SS_UDT_CATALOG_NAME, String.class);
+        getColumnMetaDataClass.put(SS_UDT_SCHEMA_NAME, String.class);
+        getColumnMetaDataClass.put(SS_UDT_ASSEMBLY_TYPE_NAME, String.class);
+        getColumnMetaDataClass.put(SS_XML_SCHEMACOLLECTION_CATALOG_NAME, String.class);
+        getColumnMetaDataClass.put(SS_XML_SCHEMACOLLECTION_SCHEMA_NAME, String.class);
+        getColumnMetaDataClass.put(SS_XML_SCHEMACOLLECTION_NAME, String.class);
+
+        try (Connection conn = getConnection()) {
+            ResultSetMetaData metadata = conn.getMetaData().getColumns(null, null, tableName, null).getMetaData();
+
+            // Ensure that there is an expected class for every column in the metadata result set
+            assertEquals(metadata.getColumnCount(), getColumnMetaDataClass.size());
+
+            for (int i = 1; i < metadata.getColumnCount(); i++) {
+                String columnLabel = metadata.getColumnLabel(i);
+                String columnClassName = metadata.getColumnClassName(i);
+                Class<?> expectedClass = getColumnMetaDataClass.get(columnLabel);
+
+                // Ensure the metadata column is in the metadata column class map
+                if (expectedClass == null) {
+                    MessageFormat form1 = new MessageFormat(TestResource.getResource("R_objectNullOrEmpty"));
+                    Object[] msgArgs1 = {"expected metadata column class for column " + columnLabel};
+                    fail(form1.format(msgArgs1));
+                }
+
+                // Ensure the actual and expected column metadata types match
+                if (!columnClassName.equals(expectedClass.getName())) {
+                    MessageFormat form1 = new MessageFormat(
+                            TestResource.getResource("R_expectedClassDoesNotMatchActualClass"));
+                    Object[] msgArgs1 = {expectedClass.getName(), columnClassName, columnLabel};
+                    fail(form1.format(msgArgs1));
+                }
+            }
+        }
+    }
+
     @BeforeAll
     public static void setupTable() throws Exception {
         setConnection();
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/datatypes/DataTypesTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/datatypes/DataTypesTest.java
index 350f7cfc..445d23dd 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/datatypes/DataTypesTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/datatypes/DataTypesTest.java
@@ -1935,6 +1935,17 @@ public class DataTypesTest extends AbstractTest {
         }
     }
 
+    @Test
+    public void testDateTimeOffsetValueOfOffsetDateTime() throws Exception {
+        OffsetDateTime expected = OffsetDateTime.now().withSecond(58).withNano(0);
+        OffsetDateTime roundUp = expected.withSecond(57).withNano(999999950);
+        OffsetDateTime roundDown = expected.withSecond(58).withNano(49);
+
+        assertEquals(expected, DateTimeOffset.valueOf(expected).getOffsetDateTime());
+        assertEquals(expected, DateTimeOffset.valueOf(roundUp).getOffsetDateTime());
+        assertEquals(expected, DateTimeOffset.valueOf(roundDown).getOffsetDateTime());
+    }
+
     static LocalDateTime getUnstorableValue() throws Exception {
         ZoneId systemTimezone = ZoneId.systemDefault();
         Instant now = Instant.now();
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/exception/ChainedExceptionTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/exception/ChainedExceptionTest.java
new file mode 100644
index 00000000..4c05d2fb
--- /dev/null
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/exception/ChainedExceptionTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.exception;
+
+import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+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.TestResource;
+import com.microsoft.sqlserver.testframework.AbstractTest;
+import com.microsoft.sqlserver.testframework.Constants;
+
+
+@RunWith(JUnitPlatform.class)
+public class ChainedExceptionTest extends AbstractTest {
+
+    @BeforeAll
+    public static void setupTests() throws Exception {
+        setConnection();
+    }
+
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    public void testTwoExceptions() throws Exception {
+        try (Connection conn = getConnection()) {
+
+            // The below should yield the following Server Messages:
+            // 1 : Msg 5074, Level 16, State 1: The object 'DF__#chained_exc__c1__AB25243A' is dependent on column 'c1'.
+            // 1 : Msg 4922, Level 16, State 9: ALTER TABLE ALTER COLUMN c1 failed because one or more objects access this column.
+            try (Statement stmnt = conn.createStatement()) {
+                stmnt.executeUpdate("CREATE TABLE #chained_exception_test_x1(c1 INT DEFAULT(0))");
+                stmnt.executeUpdate("ALTER  TABLE #chained_exception_test_x1 ALTER COLUMN c1 VARCHAR(10)");
+                stmnt.executeUpdate("DROP   TABLE IF EXISTS #chained_exception_test_x1");
+            }
+
+            fail(TestResource.getResource("R_expectedFailPassed"));
+
+        } catch (SQLException ex) {
+
+            // Check the SQLException and the chain
+            int exCount = 0;
+            int firstMsgNum = ex.getErrorCode();
+            int lastMsgNum = -1;
+
+            while (ex != null) {
+                exCount++;
+
+                lastMsgNum = ex.getErrorCode();
+
+                ex = ex.getNextException();
+            }
+
+            // Exception Count should be: 2
+            assertEquals(2, exCount, "Number of SQLExceptions in the SQLException chain");
+
+            // Check first Msg: 5074
+            assertEquals(5074, firstMsgNum, "First SQL Server Message");
+
+            // Check last Msg: 4922
+            assertEquals(4922, lastMsgNum, "Last SQL Server Message");
+        }
+    }
+}
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/fedauth/FedauthTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/fedauth/FedauthTest.java
index 4bae199d..e68fcacd 100644
--- a/src/test/java/com/microsoft/sqlserver/jdbc/fedauth/FedauthTest.java
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/fedauth/FedauthTest.java
@@ -18,6 +18,7 @@ import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.text.MessageFormat;
 import java.util.Collections;
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
@@ -323,6 +324,66 @@ public class FedauthTest extends FedauthCommon {
         }
     }
 
+    @Test
+    public void testAADServicePrincipalAuthFailureOnSubsequentConnectionsWithInvalidatedTokenCacheWithInvalidSecret() throws Exception {
+        String url = "jdbc:sqlserver://" + azureServer + ";database=" + azureDatabase + ";authentication="
+                + SqlAuthentication.ActiveDirectoryServicePrincipal + ";Username=" + applicationClientID + ";Password="
+                + applicationKey;
+
+        String invalidSecretUrl = "jdbc:sqlserver://" + azureServer + ";database=" + azureDatabase + ";authentication="
+                + SqlAuthentication.ActiveDirectoryServicePrincipal + ";Username=" + applicationClientID + ";Password="
+                + "invalidSecret";
+
+        // Should succeed on valid secret
+        try (Connection connection = DriverManager.getConnection(url)) {}
+
+        // Should fail on invalid secret
+        try (Connection connection = DriverManager.getConnection(invalidSecretUrl)) {
+            fail(TestResource.getResource("R_expectedFailPassed"));
+        } catch (Exception e) {
+            assertTrue(e.getMessage().contains(TestResource.getResource("R_invalidClientSecret")),
+                    "Expected R_invalidClientSecret error.");
+        }
+    }
+
+    @Test
+    public void testActiveDirectoryPasswordFailureOnSubsequentConnectionsWithInvalidatedTokenCacheWithInvalidPassword() throws Exception {
+
+        // Should succeed on valid password
+        try (Connection conn = DriverManager.getConnection(adPasswordConnectionStr)) {}
+
+        // Should fail on invalid password
+        try (Connection conn = DriverManager.getConnection(adPasswordConnectionStr + ";password=invalidPassword;")) {
+            fail(TestResource.getResource("R_expectedFailPassed"));
+        } catch (Exception e) {
+            MessageFormat form = new MessageFormat(TestResource.getResource("R_invalidAADAuth"));
+            Object[] msgArgs = {azureUserName, "ActiveDirectoryPassword"};
+            assertTrue(e.getMessage().contains(form.format(msgArgs)), "Expected R_invalidAADAuth error.");
+        }
+    }
+
+    @Test
+    public void testAADServicePrincipalCertAuthFailureOnSubsequentConnectionsWithInvalidatedTokenCacheWithInvalidPassword() throws Exception {
+        // Should succeed on valid cert field values
+        String url = "jdbc:sqlserver://" + azureServer + ";database=" + azureDatabase + ";authentication="
+                + SqlAuthentication.ActiveDirectoryServicePrincipalCertificate + ";Username=" + applicationClientID
+                + ";password=" + certificatePassword + ";clientCertificate=" + clientCertificate;
+
+        // Should fail on invalid cert field values
+        String invalidPasswordUrl = "jdbc:sqlserver://" + azureServer + ";database=" + azureDatabase
+                + ";authentication=" + SqlAuthentication.ActiveDirectoryServicePrincipalCertificate + ";Username="
+                + applicationClientID + ";password=invalidPassword;clientCertificate=" + clientCertificate;
+
+        try (Connection conn = DriverManager.getConnection(url)) {}
+
+        try (Connection conn = DriverManager.getConnection(invalidPasswordUrl)) {
+            fail(TestResource.getResource("R_expectedFailPassed"));
+        } catch (Exception e) {
+            assertTrue(e.getMessage().contains(TestResource.getResource("R_invalidCertFields")),
+                    "Expected R_invalidCertFields error.");
+        }
+    }
+
     /**
      * Test invalid connection property combinations when using AAD Service Principal Authentication.
      */
diff --git a/src/test/java/com/microsoft/sqlserver/jdbc/messageHandler/MessageHandlerTest.java b/src/test/java/com/microsoft/sqlserver/jdbc/messageHandler/MessageHandlerTest.java
new file mode 100644
index 00000000..b9d52a68
--- /dev/null
+++ b/src/test/java/com/microsoft/sqlserver/jdbc/messageHandler/MessageHandlerTest.java
@@ -0,0 +1,474 @@
+/*
+ * 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.messageHandler;
+
+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.assertNull;
+import org.junit.jupiter.api.Tag;
+
+import java.sql.CallableStatement;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.Statement;
+import java.sql.Types;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+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.jdbc.ISQLServerMessage;
+import com.microsoft.sqlserver.jdbc.ISQLServerMessageHandler;
+import com.microsoft.sqlserver.jdbc.SQLServerConnection;
+import com.microsoft.sqlserver.jdbc.TestResource;
+import com.microsoft.sqlserver.testframework.AbstractTest;
+import com.microsoft.sqlserver.testframework.Constants;
+
+
+@RunWith(JUnitPlatform.class)
+public class MessageHandlerTest extends AbstractTest {
+
+    @BeforeAll
+    public static void setupTests() throws Exception {
+        setConnection();
+    }
+
+    /**
+     * Helper method to count number of SQLWarnings in a chain
+     * 
+     * @param str
+     *        - Debug String, so we can evaluate from where we called it...
+     * @param sqlw
+     *        - The SQL Warning chain. (can be null)
+     * @return A count of warnings
+     */
+    private static int getWarningCount(String str, SQLWarning sqlw) {
+        int count = 0;
+        while (sqlw != null) {
+            count++;
+            // System.out.println("DEBUG: getWarningCount(): [" + str + "] SQLWarning: Error=" + sqlw.getErrorCode() + ", Severity=" + ((SQLServerWarning)sqlw).getSQLServerError().getErrorSeverity() + ", Text=|" + sqlw.getMessage() + "|.");
+            sqlw = sqlw.getNextWarning();
+        }
+        return count;
+    }
+
+    /**
+     * Test message handler with normal Statement
+     * <ul>
+     * <li>Insert duplicate row -- Mapped to Info Message</li>
+     * <li>Drop table that do not exist -- Mapped to ignore</li>
+     * </ul>
+     */
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    public void testMsgHandlerWithStatement() throws Exception {
+        try (SQLServerConnection conn = getConnection()) {
+
+            class TestMsgHandler implements ISQLServerMessageHandler {
+                int numOfCalls = 0;
+                int numOfDowngrades = 0;
+                int numOfDiscards = 0;
+
+                @Override
+                public ISQLServerMessage messageHandler(ISQLServerMessage srvErrorOrWarning) {
+                    numOfCalls++;
+                    ISQLServerMessage retObj = srvErrorOrWarning;
+
+                    if (srvErrorOrWarning.isErrorMessage()) {
+
+                        // Downgrade: 2601 -- Cannot insert duplicate key row in object 'dbo.#msghandler_tmp_table' with unique index 'ix_id'. The duplicate key value is (1)
+                        if (2601 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = srvErrorOrWarning.getSQLServerMessage().toSQLServerInfoMessage();
+                            numOfDowngrades++;
+                        }
+
+                        // Discard: 3701 -- Cannot drop the table '#msghandler_tmp_table', because it does not exist or you do not have permission.
+                        if (3701 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = null;
+                            numOfDiscards++;
+                        }
+                    }
+
+                    if (srvErrorOrWarning.isInfoMessage()) {
+
+                        // Discard: 3621 -- The statement has been terminated.
+                        if (3621 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = null;
+                            numOfDiscards++;
+                        }
+                    }
+
+                    return retObj;
+                }
+            }
+            TestMsgHandler testMsgHandler = new TestMsgHandler();
+
+            // Create a massage handler
+            conn.setServerMessageHandler(testMsgHandler);
+
+            try (Statement stmnt = conn.createStatement()) {
+
+                stmnt.executeUpdate("CREATE TABLE #msghandler_tmp_table(id int, c1 varchar(255))");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'create table', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'create table', at Statement.");
+
+                stmnt.executeUpdate("CREATE UNIQUE INDEX ix_id ON #msghandler_tmp_table(id)");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'create index', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'create index', at Statement.");
+
+                stmnt.executeUpdate("INSERT INTO #msghandler_tmp_table VALUES(1, 'row 1')");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'first insert', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'first insert', at Statement.");
+
+                stmnt.executeUpdate(
+                        "INSERT INTO #msghandler_tmp_table VALUES(1, 'row 1 - again - msg handler downgrades it')");
+                assertNotNull(conn.getWarnings(),
+                        "Expecting at least ONE SQLWarnings from 'second insert', which is a duplicate row, at Connection.");
+                assertNull(stmnt.getWarnings(),
+                        "Expecting NO SQLWarnings from 'second insert', which is a duplicate row, at Statement.");
+                conn.clearWarnings(); // Clear Warnings at Connection level
+
+                stmnt.executeUpdate("DROP TABLE #msghandler_tmp_table");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'first drop table', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'first drop table', at Statement.");
+
+                stmnt.executeUpdate("DROP TABLE #msghandler_tmp_table"); // This should be IGNORED by the message handler
+                assertNull(conn.getWarnings(),
+                        "Expecting NO SQLWarnings from 'second drop table, since it should be IGNORED by the message handler', at Connection.");
+                assertNull(stmnt.getWarnings(),
+                        "Expecting NO SQLWarnings from 'second drop table, since it should be IGNORED by the message handler', at Statement.");
+
+                // numOfCalls to the message handler should be: 3
+                assertEquals(3, testMsgHandler.numOfCalls, "Number of message calls to the message handler.");
+
+                // numOfDowngrades in the message handler should be: 1
+                assertEquals(1, testMsgHandler.numOfDowngrades, "Number of message Downgrades in the message handler.");
+
+                // numOfDiscards in the message handler should be: 2
+                assertEquals(2, testMsgHandler.numOfDiscards, "Number of message Discards in the message handler.");
+            }
+
+        } catch (SQLException ex) {
+            fail(TestResource.getResource("R_unexpectedErrorMessage") + ex.getMessage());
+        }
+    }
+
+    /**
+     * Test message handler with PreparedStatement
+     * <ul>
+     * <li>Insert duplicate row -- Mapped to Info Message</li>
+     * <li>Drop table that do not exist -- Mapped to ignore</li>
+     * </ul>
+     */
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    public void testMsgHandlerWithPreparedStatement() throws Exception {
+        try (SQLServerConnection conn = getConnection()) {
+
+            class TestMsgHandler implements ISQLServerMessageHandler {
+                int numOfCalls = 0;
+                int numOfDowngrades = 0;
+                int numOfDiscards = 0;
+
+                @Override
+                public ISQLServerMessage messageHandler(ISQLServerMessage srvErrorOrWarning) {
+                    numOfCalls++;
+                    ISQLServerMessage retObj = srvErrorOrWarning;
+
+                    if (srvErrorOrWarning.isErrorMessage()) {
+
+                        // Downgrade: 2601 -- Cannot insert duplicate key row in object 'dbo.#msghandler_tmp_table' with unique index 'ix_id'. The duplicate key value is (1)
+                        if (2601 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = srvErrorOrWarning.getSQLServerMessage().toSQLServerInfoMessage();
+                            numOfDowngrades++;
+                        }
+
+                        // Discard: 3701 -- Cannot drop the table '#msghandler_tmp_table', because it does not exist or you do not have permission.
+                        if (3701 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = null;
+                            numOfDiscards++;
+                        }
+                    }
+
+                    if (srvErrorOrWarning.isInfoMessage()) {
+
+                        // Discard: 3621 -- The statement has been terminated.
+                        if (3621 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = null;
+                            numOfDiscards++;
+                        }
+                    }
+
+                    return retObj;
+                }
+            }
+            TestMsgHandler testMsgHandler = new TestMsgHandler();
+
+            // Create a massage handler
+            conn.setServerMessageHandler(testMsgHandler);
+
+            try (Statement stmnt = conn.createStatement()) {
+                stmnt.executeUpdate("CREATE TABLE #msghandler_tmp_table(id int, c1 varchar(255))");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'create table', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'create table', at Statement.");
+            }
+
+            try (Statement stmnt = conn.createStatement()) {
+                stmnt.executeUpdate("CREATE UNIQUE INDEX ix_id ON #msghandler_tmp_table(id)");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'create index', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'create index', at Statement.");
+            }
+
+            try (PreparedStatement stmnt = conn.prepareStatement("INSERT INTO #msghandler_tmp_table VALUES(?, ?)")) {
+                stmnt.setInt(1, 1);
+                stmnt.setString(2, "row 1");
+                stmnt.executeUpdate();
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'first insert', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'first insert', at Statement.");
+            }
+
+            try (PreparedStatement stmnt = conn.prepareStatement("INSERT INTO #msghandler_tmp_table VALUES(?, ?)")) {
+                stmnt.setInt(1, 1);
+                stmnt.setString(2, "row 1 - again - msg handler downgrades it");
+                stmnt.executeUpdate();
+                assertNotNull(conn.getWarnings(),
+                        "Expecting at least ONE SQLWarnings from 'second insert', which is a duplicate row, at Connection.");
+                assertNull(stmnt.getWarnings(),
+                        "Expecting NO SQLWarnings from 'second insert', which is a duplicate row, at Statement.");
+                conn.clearWarnings(); // Clear Warnings at Connection level
+            }
+
+            try (Statement stmnt = conn.createStatement()) {
+                stmnt.executeUpdate("DROP TABLE #msghandler_tmp_table");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'first drop table', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'first drop table', at Statement.");
+            }
+
+            try (Statement stmnt = conn.createStatement()) {
+                stmnt.executeUpdate("DROP TABLE #msghandler_tmp_table");
+                assertNull(conn.getWarnings(), "Expecting NO SQLWarnings from 'first drop table', at Connection.");
+                assertNull(stmnt.getWarnings(), "Expecting NO SQLWarnings from 'first drop table', at Statement.");
+            }
+
+            // numOfCalls to the message handler should be: 3
+            assertEquals(3, testMsgHandler.numOfCalls, "Number of message calls to the message handler.");
+
+            // numOfDowngrades in the message handler should be: 1
+            assertEquals(1, testMsgHandler.numOfDowngrades, "Number of message Downgrades in the message handler.");
+
+            // numOfDiscards in the message handler should be: 2
+            assertEquals(2, testMsgHandler.numOfDiscards, "Number of message Discards in the message handler.");
+
+        } catch (SQLException ex) {
+            fail(TestResource.getResource("R_unexpectedErrorMessage") + ex.getMessage());
+        }
+    }
+
+    /**
+     * Test message handler with CallableStatement
+     * <ul>
+     * <li>Insert duplicate row -- Mapped to Info Message</li>
+     * <li>Drop table that do not exist -- Mapped to ignore</li>
+     * </ul>
+     */
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    public void testMsgHandlerWithCallableStatement() throws Exception {
+        try (SQLServerConnection conn = getConnection()) {
+
+            class TestMsgHandler implements ISQLServerMessageHandler {
+                int numOfCalls = 0;
+                int numOfDowngrades = 0;
+                int numOfDiscards = 0;
+
+                @Override
+                public ISQLServerMessage messageHandler(ISQLServerMessage srvErrorOrWarning) {
+                    numOfCalls++;
+                    ISQLServerMessage retObj = srvErrorOrWarning;
+
+                    if (srvErrorOrWarning.isErrorMessage()) {
+
+                        // Downgrade: 2601 -- Cannot insert duplicate key row in object 'dbo.#msghandler_tmp_table' with unique index 'ix_id'. The duplicate key value is (1)
+                        if (2601 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = srvErrorOrWarning.getSQLServerMessage().toSQLServerInfoMessage();
+                            numOfDowngrades++;
+                        }
+
+                        // Downgrade: 3701 -- Cannot drop the table '#msghandler_tmp_table', because it does not exist or you do not have permission.
+                        if (3701 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = null;
+                            numOfDiscards++;
+                        }
+                    }
+
+                    if (srvErrorOrWarning.isInfoMessage()) {
+
+                        // Discard: 3621 -- The statement has been terminated.
+                        if (3621 == srvErrorOrWarning.getErrorNumber()) {
+                            retObj = null;
+                            numOfDiscards++;
+                        }
+                    }
+
+                    return retObj;
+                }
+            }
+            TestMsgHandler testMsgHandler = new TestMsgHandler();
+
+            // Create a massage handler
+            conn.setServerMessageHandler(testMsgHandler);
+
+            // SQL to create procedure
+            String sqlCreateProc = "" + "CREATE PROCEDURE #msghandler_tmp_proc( \n" + "    @out_row_count INT OUTPUT \n"
+                    + ") \n" + "AS \n" + "BEGIN \n" + "    -- Create a dummy table, with index \n"
+                    + "    CREATE TABLE #msghandler_tmp_table(id int, c1 varchar(255)) \n"
+                    + "    CREATE UNIQUE INDEX ix_id ON #msghandler_tmp_table(id) \n" + "     \n"
+                    + "    -- Insert records 1 \n" + "    INSERT INTO #msghandler_tmp_table VALUES(1, 'row 1') \n"
+                    + "     \n"
+                    + "    -- Insert records 1 -- Again, which will FAIL, but the message handler will downgrade it into a INFO Message \n"
+                    + "    INSERT INTO #msghandler_tmp_table VALUES(1, 'row 1 - again - msg handler downgrades it') \n"
+                    + "     \n" + "    -- Count records \n" + "    SELECT @out_row_count = count(*) \n"
+                    + "    FROM #msghandler_tmp_table \n" + "     \n" + "    -- Drop the table \n"
+                    + "    DROP TABLE #msghandler_tmp_table \n" + "     \n"
+                    + "    -- Drop the table agin... The message handler will DISCARD the error\n"
+                    + "    DROP TABLE #msghandler_tmp_table \n" + "     \n" + "    RETURN 1 \n" + "END \n";
+
+            // Create the proc
+            try (Statement stmnt = conn.createStatement()) {
+                stmnt.executeUpdate(sqlCreateProc);
+                assertEquals(0, getWarningCount("Conn", conn.getWarnings()),
+                        "Expecting NO SQLWarnings from 'create proc', at Connection.");
+                assertEquals(0, getWarningCount("Stmnt", conn.getWarnings()),
+                        "Expecting NO SQLWarnings from 'create proc', at Statement.");
+            }
+
+            // Execute the proc
+            try (CallableStatement cstmnt = conn.prepareCall("{ ? =call  #msghandler_tmp_proc(?) }")) {
+                cstmnt.registerOutParameter(1, Types.INTEGER);
+                cstmnt.registerOutParameter(2, Types.INTEGER);
+                cstmnt.execute();
+                int procReturnCode = cstmnt.getInt(1);
+                int procRowCount = cstmnt.getInt(2);
+
+                assertEquals(1, procReturnCode, "Expecting ReturnCode 1 from the temp procedure.");
+                assertEquals(1, procRowCount, "Expecting procRowCount 1 from the temp procedure.");
+
+                assertEquals(1, getWarningCount("Conn", conn.getWarnings()),
+                        "Expecting NO SQLWarnings from 'exec proc', at Connection.");
+                assertEquals(0, getWarningCount("CStmnt", cstmnt.getWarnings()),
+                        "Expecting NO SQLWarnings from 'exec proc', at CallableStatement.");
+            }
+
+            // numOfCalls to the message handler should be: 3
+            assertEquals(3, testMsgHandler.numOfCalls, "Number of message calls to the message handler.");
+
+            // numOfDowngrades in the message handler should be: 1
+            assertEquals(1, testMsgHandler.numOfDowngrades, "Number of message Downgrades in the message handler.");
+
+            // numOfDiscards in the message handler should be: 2
+            assertEquals(2, testMsgHandler.numOfDiscards, "Number of message Discards in the message handler.");
+
+        } catch (SQLException ex) {
+            fail(TestResource.getResource("R_unexpectedErrorMessage") + ex.getMessage());
+        }
+    }
+
+    /**
+     * Test message handler with CallableStatement -- and "feedback" messages
+     * <p>
+     * Do a "long running procedure" and check that the message handler receives the "feedback" messages
+     */
+    @Test
+    @Tag(Constants.xAzureSQLDW)
+    public void testMsgHandlerWithProcedureFeedback() throws Exception {
+        try (SQLServerConnection conn = getConnection()) {
+
+            class TestMsgHandler implements ISQLServerMessageHandler {
+                int numOfCalls = 0;
+                Map<String, Long> feedbackMsgTs = new LinkedHashMap<>();
+
+                @Override
+                public ISQLServerMessage messageHandler(ISQLServerMessage srvErrorOrWarning) {
+                    numOfCalls++;
+
+                    if (50_000 == srvErrorOrWarning.getErrorNumber()) {
+                        // System.out.println("DEBUG: testMsgHandlerWithProcedureFeedback.messageHandler(): FEEDBACK: " + srvErrorOrWarning.getErrorMessage());
+                        // Remember when the message was received
+                        feedbackMsgTs.put(srvErrorOrWarning.getErrorMessage(), System.currentTimeMillis());
+                    }
+
+                    return srvErrorOrWarning;
+                }
+            }
+            TestMsgHandler testMsgHandler = new TestMsgHandler();
+
+            // Create a massage handler
+            conn.setServerMessageHandler(testMsgHandler);
+
+            int doSqlLoopCount = 4;
+            // SQL to create procedure
+            String sqlCreateProc = "" + "CREATE PROCEDURE #msghandler_feeback_proc \n" + "AS \n" + "BEGIN \n"
+                    + "    DECLARE @loop_cnt INT = " + doSqlLoopCount + " \n" + "    DECLARE @feedback VARCHAR(255) \n"
+                    + " \n" + "    WHILE (@loop_cnt > 0) \n" + "    BEGIN \n" + "        WAITFOR DELAY '00:00:01' \n"
+                    + " \n"
+                    + "        SET @feedback = 'In proc, still looping... waiting for loop_count to reach 0. loop_count is now at: ' + convert(varchar(10), @loop_cnt) \n"
+                    + "        RAISERROR(@feedback, 0, 1) WITH NOWAIT \n" + " \n"
+                    + "        SET @loop_cnt = @loop_cnt - 1 \n" + "    END \n" + " \n" + "    RETURN @loop_cnt \n"
+                    + "END \n";
+
+            // Create the proc
+            try (Statement stmnt = conn.createStatement()) {
+                stmnt.executeUpdate(sqlCreateProc);
+                assertEquals(0, getWarningCount("Conn", conn.getWarnings()),
+                        "Expecting NO SQLWarnings from 'create proc', at Connection.");
+                assertEquals(0, getWarningCount("Stmnt", conn.getWarnings()),
+                        "Expecting NO SQLWarnings from 'create proc', at Statement.");
+            }
+
+            // Execute the proc
+            try (CallableStatement cstmnt = conn.prepareCall("{ ? =call  #msghandler_feeback_proc }")) {
+                cstmnt.registerOutParameter(1, Types.INTEGER);
+                cstmnt.execute();
+                int procReturnCode = cstmnt.getInt(1);
+
+                assertEquals(0, procReturnCode, "Unexpected ReturnCode from the temp procedure.");
+
+                assertEquals(0, getWarningCount("conn", conn.getWarnings()),
+                        "Unexpected Number Of SQLWarnings from 'exec proc', at Connection.");
+                assertEquals(doSqlLoopCount, getWarningCount("cstmnt", cstmnt.getWarnings()),
+                        "Unexpected Number Of SQLWarnings from 'exec proc', at CallableStatement.");
+            }
+
+            // numOfCalls to the message handler should be: #
+            assertEquals(doSqlLoopCount, testMsgHandler.numOfCalls, "Number of message calls to the message handler.");
+
+            // Loop all received messages and check that they are within a second (+-200ms)
+            long prevTime = 0;
+            for (Entry<String, Long> entry : testMsgHandler.feedbackMsgTs.entrySet()) {
+                if (prevTime == 0) {
+                    prevTime = entry.getValue();
+                    continue;
+                }
+
+                long msDiff = entry.getValue() - prevTime;
+                if (msDiff < 800 || msDiff > 1200) {
+                    fail("Received Messages is to far apart. They should be approx 1000 ms. msDiff=" + msDiff
+                            + " Message=|" + entry.getKey() + "|.");
+                }
+
+                prevTime = entry.getValue();
+            }
+
+        } catch (SQLException ex) {
+            fail(TestResource.getResource("R_unexpectedErrorMessage") + ex.getMessage());
+        }
+    }
+}
diff --git a/src/test/java/com/microsoft/sqlserver/testframework/AbstractTest.java b/src/test/java/com/microsoft/sqlserver/testframework/AbstractTest.java
index d17fa146..1ce9cd19 100644
--- a/src/test/java/com/microsoft/sqlserver/testframework/AbstractTest.java
+++ b/src/test/java/com/microsoft/sqlserver/testframework/AbstractTest.java
@@ -62,6 +62,9 @@ public abstract class AbstractTest {
     protected static String tenantID;
     protected static String[] keyIDs = null;
 
+    protected static String linkedServer = null;
+    protected static String remoteUser = null;
+    protected static String remotePassword = null;
     protected static String[] enclaveServer = null;
     protected static String[] enclaveAttestationUrl = null;
     protected static String[] enclaveAttestationProtocol = null;
@@ -197,6 +200,10 @@ public abstract class AbstractTest {
 
         clientKeyPassword = getConfiguredProperty("clientKeyPassword", "");
 
+        linkedServer = getConfiguredProperty("linkedServer", null);
+        remoteUser = getConfiguredProperty("remoteUser", null);
+        remotePassword = getConfiguredProperty("remotePassword", null);
+
         kerberosServer = getConfiguredProperty("kerberosServer", null);
         kerberosServerPort = getConfiguredProperty("kerberosServerPort", null);
 
diff --git a/src/test/resources/BulkCopyCSVTestInputDelimiterEscape.csv b/src/test/resources/BulkCopyCSVTestInputDelimiterEscape.csv
index 10afd5d8..9425b0a1 100644
--- a/src/test/resources/BulkCopyCSVTestInputDelimiterEscape.csv
+++ b/src/test/resources/BulkCopyCSVTestInputDelimiterEscape.csv
@@ -9,3 +9,7 @@
 9@1997@Ford@E350@"Super@ ""luxurious"" truck"@ 
 10@1997@ "Ford" @E350@  "E63"@ 
 11@1997@Ford@E350@" Super luxurious truck "@ 
+12@1997@"F
+o
+r
+d"@"E350"@"""Super"" ""luxurious"" ""truck"""@