Give each redshift client their own boto session

CHANGELOG.md

@@ -3,8 +3,12 @@
 ### Feature
 - Added 'Last Modified' stat in snowflake catalog macro. Now should be available in docs. ([#2728](https://github.com/fishtown-analytics/dbt/issues/2728))
 
+### Fixes
+- `dbt compile` and `dbt run` failed with `KeyError: 'endpoint_resolver'` when threads > 1 and `method: iam` had been specified in the profiles.yaml ([#2756](https://github.com/fishtown-analytics/dbt/issues/2756), [#2766](https://github.com/fishtown-analytics/dbt/pull/2766))
+
 Contributors:
 - [@Mr-Nobody99](https://github.com/Mr-Nobody99) ([#2732](https://github.com/fishtown-analytics/dbt/pull/2732))
+- [@jweibel22](https://github.com/jweibel22) ([#2766](https://github.com/fishtown-analytics/dbt/pull/2766))
 
 ## dbt 0.18.1b2 (September 22, 2020)
 

plugins/redshift/dbt/adapters/redshift/connections.py

@@ -99,7 +99,8 @@ def fetch_cluster_credentials(cls, db_user, db_name, cluster_id,
         must already exist in the database, or else an error will occur"""
 
         if iam_profile is None:
-            boto_client = boto3.client('redshift')
+            session = boto3.Session()
+            boto_client = session.client("redshift")
         else:
             logger.debug("Connecting to Redshift using 'IAM'" +
                          f"with profile {iam_profile}")

test/unit/test_redshift_adapter.py

@@ -1,6 +1,9 @@
 import unittest
 from unittest import mock
+from unittest.mock import Mock
+
 import agate
+import boto3
 
 import dbt.adapters  # noqa
 import dbt.flags as flags
@@ -128,6 +131,23 @@ def test_invalid_iam_no_cluster_id(self):
 
         self.assertTrue("'cluster_id' must be provided" in context.exception.msg)
 
+    def test_default_session_is_not_used_when_iam_used(self):
+        boto3.DEFAULT_SESSION = Mock()
+        self.config.credentials = self.config.credentials.replace(method='iam')
+        self.config.credentials.cluster_id = 'clusterid'
+        with mock.patch('dbt.adapters.redshift.connections.boto3.Session'):
+            RedshiftAdapter.ConnectionManager.get_credentials(self.config.credentials)
+            self.assertEquals(boto3.DEFAULT_SESSION.client.call_count, 0,
+                              "The redshift client should not be created using the default session because the session object is not thread-safe")
+
+    def test_default_session_is_not_used_when_iam_not_used(self):
+        boto3.DEFAULT_SESSION = Mock()
+        self.config.credentials = self.config.credentials.replace(method=None)
+        with mock.patch('dbt.adapters.redshift.connections.boto3.Session'):
+            RedshiftAdapter.ConnectionManager.get_credentials(self.config.credentials)
+            self.assertEquals(boto3.DEFAULT_SESSION.client.call_count, 0,
+                              "The redshift client should not be created using the default session because the session object is not thread-safe")
+
     def test_cancel_open_connections_empty(self):
         self.assertEqual(len(list(self.adapter.cancel_open_connections())), 0)