@ Override public Collection < Symptom > detect ( Collection < Measurement > measurements ) { publishingMetrics . executeDetectorIncr ( BACK_PRESSURE_DETECTOR ) ; Collection < Symptom > result = new ArrayList <> ( ) ; Instant now = context . checkpoint ( ) ; MeasurementsTable bpMetrics = MeasurementsTable . of ( measurements ) . type ( METRIC_BACK_PRESSURE . text ( ) ) ; for ( String component : bpMetrics . uniqueComponents ( ) ) { double compBackPressure = bpMetrics . component ( component ) . sum ( ) ; if ( compBackPressure > noiseFilterMillis ) { LOG . info ( String . format ( "Detected component back-pressure for %s, total back pressure is %f" , component , compBackPressure ) ) ; List < String > addresses = Collections . singletonList ( component ) ; result . add ( new Symptom ( SYMPTOM_COMP_BACK_PRESSURE . text ( ) , now , addresses ) ) ; } } for ( String instance : bpMetrics . uniqueInstances ( ) ) { double totalBP = bpMetrics . instance ( instance ) . sum ( ) ; if ( totalBP > noiseFilterMillis ) { LOG . info ( String . format ( "Detected instance back-pressure for %s, total back pressure is %f" , instance , totalBP ) ) ; List < String > addresses = Collections . singletonList ( instance ) ; result . add ( new Symptom ( SYMPTOM_INSTANCE_BACK_PRESSURE . text ( ) , now , addresses ) ) ; } } return result ; }
Detects all components initiating backpressure above the configured limit . Normally there will be only one component