add consumer_member_id tag to partitions assign / revoked metrics (#358)

Author: mwarkentin

arroyo/processing/processor.py

@@ -236,7 +236,7 @@ def on_partitions_assigned(partitions: Mapping[Partition, int]) -> None:
             logger.info("New partitions assigned: %r", partitions)
             logger.info("Member id: %r", self.__consumer.member_id)
             self.__metrics_buffer.metrics.increment(
-                "arroyo.consumer.partitions_assigned.count", len(partitions)
+                "arroyo.consumer.partitions_assigned.count", len(partitions), tags={"consumer_member_id": self.__consumer.member_id}
             )
 
             current_partitions = dict(self.__consumer.tell())
@@ -262,7 +262,7 @@ def on_partitions_revoked(partitions: Sequence[Partition]) -> None:
             logger.info("Partitions to revoke: %r", partitions)
 
             self.__metrics_buffer.metrics.increment(
-                "arroyo.consumer.partitions_revoked.count", len(partitions)
+                "arroyo.consumer.partitions_revoked.count", len(partitions), tags={"consumer_member_id": self.__consumer.member_id}
             )
 
             if partitions: