(self, consumer_offsets, **kwargs)
Report the consumer group offsets and consumer lag.
Report the consumer group offsets and consumer lag.
Report the consumer group offsets and consumer lag .
def _report_consumer_offsets_and_lag(self, consumer_offsets, **kwargs):
        """Report the consumer group offsets and consumer lag."""
        for (consumer_group, topic, partition), consumer_offset in consumer_offsets.items():
            consumer_group_tags = ['topic:%s' % topic, 'partition:%s' % partition, 'consumer_group:%s' % consumer_group]
            if 'source' in kwargs:
                consumer_group_tags.append('source:%s' % kwargs['source'])
            consumer_group_tags.extend(self._custom_tags)

            partitions = self.kafka_client.cluster.partitions_for_topic(topic)
            if partitions is not None and partition in partitions:
                # report consumer offset if the partition is valid because even if leaderless the consumer offset will
                # be valid once the leader failover completes
                self.gauge('consumer_offset', consumer_offset, tags=consumer_group_tags)

                if (topic, partition) not in self._highwater_offsets:
                    self.log.warning(
                        "Consumer group: %s has offsets for topic: %s partition: %s, but no stored highwater offset "
                        "(likely the partition is in the middle of leader failover) so cannot calculate consumer lag.",
                        consumer_group,
                        topic,
                        partition,
                    )
                    continue

                consumer_lag = self._highwater_offsets[(topic, partition)] - consumer_offset
                self.gauge('consumer_lag', consumer_lag, tags=consumer_group_tags)

                if consumer_lag < 0:  # this will effectively result in data loss, so emit an event for max visibility
                    title = "Negative consumer lag for group: {}.".format(consumer_group)
                    message = (
                        "Consumer group: {}, topic: {}, partition: {} has negative consumer lag. This should never "
                        "happen and will result in the consumer skipping new messages until the lag turns "
                        "positive.".format(consumer_group, topic, partition)
                    )
                    key = "{}:{}:{}".format(consumer_group, topic, partition)
                    self.send_event(title, message, consumer_group_tags, 'consumer_lag', key, severity="error")
                    self.log.debug(message)

            else:
                if partitions is None:
                    msg = (
                        "Consumer group: %s has offsets for topic: %s, partition: %s, but that topic has no partitions "
                        "in the cluster, so skipping reporting these offsets."
                    )
                else:
                    msg = (
                        "Consumer group: %s has offsets for topic: %s, partition: %s, but that topic partition isn't "
                        "included in the cluster partitions, so skipping reporting these offsets."
                    )
                self.log.warning(msg, consumer_group, topic, partition)
                self.kafka_client.cluster.request_update()
def _report_consumer_offsets_and_lag ( self , consumer_offsets , * * kwargs ) : for ( consumer_group , topic , partition ) , consumer_offset in consumer_offsets . items ( ) : consumer_group_tags = [ 'topic:%s' % topic , 'partition:%s' % partition , 'consumer_group:%s' % consumer_group ] if 'source' in kwargs : consumer_group_tags . append ( 'source:%s' % kwargs [ 'source' ] ) consumer_group_tags . extend ( self . _custom_tags ) partitions = self . kafka_client . cluster . partitions_for_topic ( topic ) if partitions is not None and partition in partitions : # report consumer offset if the partition is valid because even if leaderless the consumer offset will # be valid once the leader failover completes self . gauge ( 'consumer_offset' , consumer_offset , tags = consumer_group_tags ) if ( topic , partition ) not in self . _highwater_offsets : self . log . warning ( "Consumer group: %s has offsets for topic: %s partition: %s, but no stored highwater offset " "(likely the partition is in the middle of leader failover) so cannot calculate consumer lag." , consumer_group , topic , partition , ) continue consumer_lag = self . _highwater_offsets [ ( topic , partition ) ] - consumer_offset self . gauge ( 'consumer_lag' , consumer_lag , tags = consumer_group_tags ) if consumer_lag < 0 : # this will effectively result in data loss, so emit an event for max visibility title = "Negative consumer lag for group: {}." . format ( consumer_group ) message = ( "Consumer group: {}, topic: {}, partition: {} has negative consumer lag. This should never " "happen and will result in the consumer skipping new messages until the lag turns " "positive." . format ( consumer_group , topic , partition ) ) key = "{}:{}:{}" . format ( consumer_group , topic , partition ) self . send_event ( title , message , consumer_group_tags , 'consumer_lag' , key , severity = "error" ) self . log . debug ( message ) else : if partitions is None : msg = ( "Consumer group: %s has offsets for topic: %s, partition: %s, but that topic has no partitions " "in the cluster, so skipping reporting these offsets." ) else : msg = ( "Consumer group: %s has offsets for topic: %s, partition: %s, but that topic partition isn't " "included in the cluster partitions, so skipping reporting these offsets." ) self . log . warning ( msg , consumer_group , topic , partition ) self . kafka_client . cluster . request_update ( )