dekaf: Track latest committed offset by group and journal

[jshearer]

Looks like this:

# TYPE dekaf_committed_offset gauge
dekaf_committed_offset{group_id="joseph-dekaf-testing-1100-1",journal_name="demo/bots-humans-edits/pivot=00"} 73173569244
dekaf_committed_offset{group_id="joseph-dekaf-testing-1100-0",journal_name="demo/wikipedia/recentchange/pivot=00"} 2036875838205

In order to correlate this with the journal:gazette_write_head:max metric, we need the exact journal name. Unfortunately, Kafka only represents topic partitions by their index, so in order to expose a useful metric we have to go and look up the exact journal name given the collection name and partition index. Fortunately, OffsetCommit is a fairly infrequent API call, and is not expected to respond particularly quickly, so I don't see this as a huge issue. If it is, we could cache this mapping from the various places it's fetched elsewhere.

---

This change is 

[jgraettinger]

LGTM

[jgraettinger]

nit: is this clone needed?

[jshearer]

AFAICT, it's needed either here or inside authenticated_client(). authenticated_client()'s signature desugars to

pub async fn authenticated_client<'a>(&'a mut self) -> anyhow::Result<&'a flow_client::Client>

AFAICT, because the returned reference &'a flow_client::Client, has the same lifetime as the mutable reference to self, it too represents a mutable borrow. This causes problems later on in offset_commit where we call self.decrypt_topic_name which takes a &self, which is disallowed because the flow_client holds a mut reference with lifetime >= &mut self, and we get the error

cannot borrow `*self` as immutable because it is also borrowed as mutable
immutable borrow occurs here // call to self.decrypt_topic_name()
session.rs(850, 27): mutable borrow occurs here // call to self.auth.authenticated_client()
session.rs(873, 57): mutable borrow later used here // call to Collection::new(&flow_client)

I could change authenticated_client() to return anyhow::Result<flow_client::Client>, but that would just mean .clone()ing the client in there instead. Maybe that's better ergonomically?

[jgraettinger]

Nope, this is good. I just didn't recall that it already returned a reference. IMO it's better to let the caller decide whether to take a clone or use a reference, rather than taking that choice from them.