Fix two sources of SQL statement leaks. (#13259)

* Fix two sources of SQL statement leaks.

1) SqlTaskResource and DruidJdbcResultSet leaked statements 100% of the
   time, since they call stmt.plan(), which adds statements to
   SqlLifecycleManager, and they do not explicitly remove them.

2) SqlResource leaked statements if yielder.close() threw an exception.
   (And also would not emit metrics, since in that case it failed to
   call stmt.close as well.)

* Only closeQuietly is needed.

sql/src/main/java/org/apache/druid/sql/DirectStatement.java

@@ -210,7 +210,8 @@ public class DirectStatement extends AbstractStatement implements Cancelable
         // Context keys for authorization. Use the user-provided keys,
         // NOT the keys from the query context which, by this point,
         // will have been extended with internally-defined values.
-        queryPlus.context().keySet())) {
+        queryPlus.context().keySet()
+    )) {
       validate(planner);
       authorize(planner, authorizer());
 
@@ -314,4 +315,11 @@ public class DirectStatement extends AbstractStatement implements Cancelable
       state = State.CLOSED;
     }
   }
+
+  @Override
+  public void closeQuietly()
+  {
+    sqlToolbox.sqlLifecycleManager.remove(sqlQueryId(), this);
+    super.closeQuietly();
+  }
 }

sql/src/main/java/org/apache/druid/sql/avatica/DruidJdbcResultSet.java

@@ -411,7 +411,8 @@ public class DruidJdbcResultSet implements Closeable
       throw new RuntimeException(t);
     }
     finally {
-      // Closing the statement cause logs and metrics to be emitted.
+      // Closing the statement causes logs and metrics to be emitted, and causes the statement to be removed
+      // from the SqlLifecycleManager.
       stmt.close();
     }
   }

sql/src/main/java/org/apache/druid/sql/http/SqlResource.java

@@ -55,6 +55,7 @@ import org.apache.druid.sql.SqlLifecycleManager.Cancelable;
 import org.apache.druid.sql.SqlPlanningException;
 import org.apache.druid.sql.SqlRowTransformer;
 import org.apache.druid.sql.SqlStatementFactory;
+import org.apache.druid.utils.CloseableUtils;
 
 import javax.annotation.Nullable;
 import javax.servlet.http.HttpServletRequest;
@@ -174,8 +175,12 @@ public class SqlResource
                     throw new RuntimeException(ex);
                   }
                   finally {
-                    yielder.close();
+                    final Exception finalE = e;
-                    endLifecycle(stmt, e, os.getCount());
+
+                    CloseableUtils.closeAll(
+                        yielder,
+                        () -> endLifecycle(stmt, finalE, os.getCount())
+                    );
                   }
                 }
               }
@@ -245,7 +250,6 @@ public class SqlResource
       HttpStatement stmt
   )
   {
-    sqlLifecycleManager.remove(stmt.sqlQueryId(), stmt);
     stmt.closeQuietly();
   }
 
@@ -261,7 +265,6 @@ public class SqlResource
     } else {
       reporter.failed(e);
     }
-    sqlLifecycleManager.remove(stmt.sqlQueryId(), stmt);
     stmt.close();
   }