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();
    }
  }
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
Execute given query and process it s result set .