@ Override public void recordAppend ( String streamSegmentName , long dataLength , int numOfEvents , Duration elapsed ) { getWriteStreamSegment ( ) . reportSuccessEvent ( elapsed ) ; DynamicLogger dl = getDynamicLogger ( ) ; dl . incCounterValue ( globalMetricName ( SEGMENT_WRITE_BYTES ) , dataLength ) ; dl . incCounterValue ( globalMetricName ( SEGMENT_WRITE_EVENTS ) , numOfEvents ) ; if ( ! StreamSegmentNameUtils . isTransactionSegment ( streamSegmentName ) ) { //Don't report segment specific metrics if segment is a transaction //The parent segment metrics will be updated once the transaction is merged dl . incCounterValue ( SEGMENT_WRITE_BYTES , dataLength , segmentTags ( streamSegmentName ) ) ; dl . incCounterValue ( SEGMENT_WRITE_EVENTS , numOfEvents , segmentTags ( streamSegmentName ) ) ; try { SegmentAggregates aggregates = getSegmentAggregate ( streamSegmentName ) ; // Note: we could get stats for a transaction segment. We will simply ignore this as we // do not maintain intermittent txn segment stats. Txn stats will be accounted for // only upon txn commit. This is done via merge method. So here we can get a txn which // we do not know about and hence we can get null and ignore. if ( aggregates != null && aggregates . update ( dataLength , numOfEvents ) ) { report ( streamSegmentName , aggregates ) ; } } catch ( Exception e ) { log . warn ( "Record statistic for {} for data: {} and events:{} threw exception" , streamSegmentName , dataLength , numOfEvents , e ) ; } } }
Updates segment specific aggregates . Then if two minutes have elapsed between last report of aggregates for this segment send a new update to the monitor . This update to the monitor is processed by monitor asynchronously .