@Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
            	List<Metric> dequeuedMetrics = collectionService.commitMetrics(METRIC_MESSAGES_CHUNK_SIZE, TIMEOUT);

            	int noOfDatapointsCommitted=0;
             	for(Metric metric:dequeuedMetrics){
             		noOfDatapointsCommitted+=metric.getDatapoints().size();
            	}
                if (dequeuedMetrics.size() > 0) {
                    LOGGER.info(MessageFormat.format("Committed {0} metrics.", dequeuedMetrics.size()));
                    monitorService.modifyCounter(Counter.COMMIT_CLIENT_METRIC_WRITES, dequeuedMetrics.size(), new HashMap<String,String>());
                }
                if(noOfDatapointsCommitted>0){
                	LOGGER.debug(MessageFormat.format("Committed {0} datapoints.", noOfDatapointsCommitted));
                    jobCounter.addAndGet(noOfDatapointsCommitted);
                    monitorService.modifyCounter(Counter.COMMIT_CLIENT_DATAPOINT_WRITES, noOfDatapointsCommitted, new HashMap<String,String>());
                }
                
                Thread.sleep(POLL_INTERVAL_MS);
            } catch (InterruptedException ie) {
                LOGGER.info("Execution was interrupted.");
                Thread.currentThread().interrupt();
                break;
            } catch (Throwable ex) {
                LOGGER.info("Error occurred while committing metrics. Reason {}", ex.toString());
            }
        }
        LOGGER.warn(MessageFormat.format("Metric committer thread interrupted. {} datapoints committed by this thread.", jobCounter.get()));
        collectionService.dispose();
        monitorService.dispose();
    }
@ Override public void run ( ) { while ( ! Thread . currentThread ( ) . isInterrupted ( ) ) { try { List < Metric > dequeuedMetrics = collectionService . commitMetrics ( METRIC_MESSAGES_CHUNK_SIZE , TIMEOUT ) ; int noOfDatapointsCommitted = 0 ; for ( Metric metric : dequeuedMetrics ) { noOfDatapointsCommitted += metric . getDatapoints ( ) . size ( ) ; } if ( dequeuedMetrics . size ( ) > 0 ) { LOGGER . info ( MessageFormat . format ( "Committed {0} metrics." , dequeuedMetrics . size ( ) ) ) ; monitorService . modifyCounter ( Counter . COMMIT_CLIENT_METRIC_WRITES , dequeuedMetrics . size ( ) , new HashMap < String , String > ( ) ) ; } if ( noOfDatapointsCommitted > 0 ) { LOGGER . debug ( MessageFormat . format ( "Committed {0} datapoints." , noOfDatapointsCommitted ) ) ; jobCounter . addAndGet ( noOfDatapointsCommitted ) ; monitorService . modifyCounter ( Counter . COMMIT_CLIENT_DATAPOINT_WRITES , noOfDatapointsCommitted , new HashMap < String , String > ( ) ) ; } Thread . sleep ( POLL_INTERVAL_MS ) ; } catch ( InterruptedException ie ) { LOGGER . info ( "Execution was interrupted." ) ; Thread . currentThread ( ) . interrupt ( ) ; break ; } catch ( Throwable ex ) { LOGGER . info ( "Error occurred while committing metrics. Reason {}" , ex . toString ( ) ) ; } } LOGGER . warn ( MessageFormat . format ( "Metric committer thread interrupted. {} datapoints committed by this thread." , jobCounter . get ( ) ) ) ; collectionService . dispose ( ) ; monitorService . dispose ( ) ; }
~ Methods **************************************************************************************************************************************