ref(rust): Add some rebalancing-related metrics (#5167)

* ref(rust): Add some rebalancing-related metrics

Port over the parts from #310
that make sense -- some of the metrics I added there are duplicates and
are named poorly, and I don't think we need those to line up 1:1 (also
because callbacks in rust work slightly differently)

* fmt

src/processing/mod.rs

@@ -88,13 +88,35 @@ impl<TPayload: 'static> AssignmentCallbacks for Callbacks<TPayload> {
     // Revisit this so that it is not the callback that perform the
     // initialization.  But we just provide a signal back to the
     // processor to do that.
-    fn on_assign(&self, _: HashMap<Partition, u64>) {
+    fn on_assign(&self, partitions: HashMap<Partition, u64>) {
+        let metrics = get_metrics();
+        metrics.increment(
+            "arroyo.consumer.partitions_assigned.count",
+            partitions.len() as i64,
+            None,
+        );
+
+        let start = Instant::now();
+
         let mut state = self.0.lock().unwrap();
         state.strategy = Some(state.processing_factory.create());
+
+        metrics.timing(
+            "arroyo.consumer.create_strategy.time",
+            start.elapsed().as_millis() as u64,
+            None,
+        );
     }
-    fn on_revoke<C: CommitOffsets>(&self, commit_offsets: C, _: Vec<Partition>) {
+
+    fn on_revoke<C: CommitOffsets>(&self, commit_offsets: C, partitions: Vec<Partition>) {
         tracing::info!("Start revoke partitions");
         let metrics = get_metrics();
+        metrics.increment(
+            "arroyo.consumer.partitions_revoked.count",
+            partitions.len() as i64,
+            None,
+        );
+
         let start = Instant::now();
 
         let mut state = self.0.lock().unwrap();