Improve handling of interrupts in TrinoResultSet

Differentiate between interrupts happening in the calling thread and
interrupts of the background thread.

- use different message
- the interrupts happening in the calling thread should result in full
  cancellation of the background thread. Previously they would close
  client, but wouldn't set `cancelled` flag.

client/trino-jdbc/src/main/java/io/trino/jdbc/TrinoResultSet.java

@@ -183,7 +183,9 @@ public AsyncIterator(Iterator<T> dataIterator, StatementClient client)
                     }
                 }
                 catch (InterruptedException e) {
-                    handleInterrupt(e);
+                    client.close();
+                    rowQueue.clear();
+                    throw new RuntimeException(new SQLException("ResultSet thread was interrupted", e));
                 }
                 finally {
                     semaphore.release();
@@ -196,18 +198,6 @@ public void cancel()
         {
             cancelled = true;
             future.cancel(true);
-            cleanup();
-        }
-
-        private void handleInterrupt(InterruptedException e)
-        {
-            cleanup();
-            Thread.currentThread().interrupt();
-            throw new RuntimeException(new SQLException("ResultSet thread was interrupted", e));
-        }
-
-        private void cleanup()
-        {
             // When thread interruption is mis-handled by underlying implementation of `client`, the thread which
             // is working for `future` may be blocked by `rowQueue.put` (`rowQueue` is full) and will never finish
             // its work. It is necessary to close `client` and drain `rowQueue` to avoid such leaks.
@@ -253,6 +243,13 @@ protected T computeNext()
             }
             return rowQueue.poll();
         }
+
+        private void handleInterrupt(InterruptedException e)
+        {
+            cancel();
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(new SQLException("Interrupted", e));
+        }
     }
 
     private static class ResultsPageIterator

client/trino-jdbc/src/test/java/io/trino/jdbc/TestTrinoDriver.java

@@ -906,7 +906,7 @@ public void testQueryCancelByInterrupt()
         assertTrue(queryFinished.await(10, SECONDS));
         assertThat(queryFailure.get())
                 .isInstanceOf(SQLException.class)
-                .hasMessage("ResultSet thread was interrupted");
+                .hasMessage("Interrupted");
         assertEquals(getQueryState(queryId.get()), FAILED);
     }