fix: Hanging session after python model is complete

CHANGELOG.md

@@ -4,6 +4,7 @@
 - This fix addresses a bug where spark.sql('use ...') would fail with a None database error when the database field wasn't set in profiles.yml, by falling back to the schema value.
 - Fixed incorrect error response assumption in Glue statement output. The `Status` field from `Statement.Output` should only return lowercase `ok` or `error`. The previous check `output.get('Status') == 'ERROR'` would miss non-uppercase, causing errors to silently pass as successful executions.
 - Fixed incremental append and partition_by support for Iceberg Python
+- Fixed hanging Glue session after a Python model completes. The session is now closed in a `finally` block after Python model execution (success or failure), unless `glue_session_reuse: true` is set, in which case `close_session()` returns early and leaves the session running.
 
 ## 1.10.19
 

dbt/adapters/glue/python_submissions.py

@@ -5,9 +5,9 @@
 
 from dbt.adapters.base import PythonJobHelper
 from dbt_common.exceptions import DbtRuntimeError
-
 from dbt.adapters.glue import GlueCredentials
 
+
 class GluePythonJobHelper(PythonJobHelper):
     def __init__(self, parsed_model: Dict, credentials: GlueCredentials) -> None:
         self.credentials = credentials
@@ -36,17 +36,20 @@ def submit(self, compiled_code: str) -> None:
         glue_client = connection.client
         session_id = connection.session_id
 
-        print(f"DEBUG: Using Glue session: {session_id}")
-
         try:
             # Run the actual Python code
             statement_id = self._run_statement(glue_client, session_id, compiled_code)
-            
+
             # Wait for completion
             self._wait_for_statement_completion(glue_client, session_id, statement_id)
-            
+
         except Exception as e:
             raise DbtRuntimeError(f"Python model execution failed: {str(e)}")
+        finally:
+            # Close the session unless glue_session_reuse is enabled.
+            # close_session() already honors glue_session_reuse and returns
+            # early without stopping the session when reuse is requested.
+            connection.close_session()
 
     def _run_statement(self, glue_client, session_id, code):
         """Run a Python statement in the existing Glue session"""
@@ -55,7 +58,7 @@ def _run_statement(self, glue_client, session_id, code):
             Code=code
         )
         return response['Id']
-    
+
     def _wait_for_statement_completion(self, glue_client, session_id, statement_id):
         """Wait for a statement to complete execution"""
         start_time = time.time()