@ Override public Collection < Measurement > fetch ( ) { Collection < Measurement > result = new ArrayList <> ( ) ; Instant now = context . checkpoint ( ) ; List < String > boltComponents = physicalPlanProvider . getBoltNames ( ) ; Duration duration = getDuration ( ) ; for ( String component : boltComponents ) { String [ ] boltInstanceNames = packingPlanProvider . getBoltInstanceNames ( component ) ; for ( String instance : boltInstanceNames ) { String metric = getMetricName ( ) + instance + MetricName . METRIC_WAIT_Q_SIZE_SUFFIX ; Collection < Measurement > stmgrResult = metricsProvider . getMeasurements ( now , duration , metric , COMPONENT_STMGR ) ; if ( stmgrResult . isEmpty ( ) ) { continue ; } MeasurementsTable table = MeasurementsTable . of ( stmgrResult ) . component ( COMPONENT_STMGR ) ; if ( table . size ( ) == 0 ) { continue ; } double totalSize = table . type ( metric ) . sum ( ) ; Measurement measurement = new Measurement ( component , instance , getMetricName ( ) , now , totalSize ) ; result . add ( measurement ) ; } } return result ; }
The buffer size as provided by tracker