hive-protolib/src/main/java/com/streamsets/pipeline/stage/lib/hive/HiveQueryExecutor.java
private<T> T executeQuery(String query, WithResultSet<T> execution) throws StageException {
    LOG.debug("Executing SQL:  {}", query);
    Timer.Context t = selectTimer.time();
    try(
      Statement statement = hiveConfigBean.getHiveConnection().createStatement();
      ResultSet rs = statement.executeQuery(query);
    ) {
      // Stop timer immediately so that we're calculating only query execution time and not the processing time
      long time = t.stop();
      LOG.debug("Query '{}' took {} nanoseconds", query, time);
      t = null;

      return execution.run(rs);
    } catch(Exception e) {
      LOG.error("Exception while processing query: {}", query, e);
      throw new HiveStageCheckedException(Errors.HIVE_20, query, e.getMessage());
    } finally {
      // If the timer wasn't stopped due to exception yet, stop it now
      if(t != null) {
        long time = t.stop();
        LOG.debug("Query '{}' took {} nanoseconds", query, time);
      }
      selectMeter.mark();
    }
  }
private<T> T executeQuery(String query, WithResultSet<T> execution) throws StageException {
    LOG.debug("Executing SQL:  {}", query);
    Timer.Context t = selectTimer.time();
    try(
      Statement statement = hiveConfigBean.getHiveConnection().createStatement();
      ResultSet rs = statement.executeQuery(query);
    ) {
      // Stop timer immediately so that we're calculating only query execution time and not the processing time
      long time = t.stop();
      LOG.debug("Query '{}' took {} nanoseconds", query, time);
      t = null;

      return execution.run(rs);
    } catch(Exception e) {
      LOG.error("Exception while processing query: {}", query, e);
      throw new HiveStageCheckedException(Errors.HIVE_20, query, e.getMessage());
    } finally {
      // If the timer wasn't stopped due to exception yet, stop it now
      if(t != null) {
        long time = t.stop();
        LOG.debug("Query '{}' took {} nanoseconds", query, time);
      }
      selectMeter.mark();
    }
  }
Execute given query and process it's result set.

With all required side effects such as catching exceptions, updating metrics, ...

@param query Query to execute
@param execution Action that should be carried away on the result set
@throws StageException