(self, contexts_limit)
Report the broker highwater offsets.
Report the broker highwater offsets.
Report the broker highwater offsets .
def _report_highwater_offsets(self, contexts_limit):
        """Report the broker highwater offsets."""
        reported_contexts = 0
        for (topic, partition), highwater_offset in self._highwater_offsets.items():
            broker_tags = ['topic:%s' % topic, 'partition:%s' % partition]
            broker_tags.extend(self._custom_tags)
            self.gauge('broker_offset', highwater_offset, tags=broker_tags)
            reported_contexts += 1
            if reported_contexts == contexts_limit:
                return
def _report_highwater_offsets ( self , contexts_limit ) : reported_contexts = 0 for ( topic , partition ) , highwater_offset in self . _highwater_offsets . items ( ) : broker_tags = [ 'topic:%s' % topic , 'partition:%s' % partition ] broker_tags . extend ( self . _custom_tags ) self . gauge ( 'broker_offset' , highwater_offset , tags = broker_tags ) reported_contexts += 1 if reported_contexts == contexts_limit : return