Capture RDKafka Statistics as an internal built-in log

src/dataflow/src/logging/materialized.rs

@@ -315,7 +315,7 @@ pub fn construct<A: Allocate>(
             move |((consumer_name, source_id, partition_id), diff_vector)| {
                 row_packer.pack(&[
                     Datum::String(&consumer_name),
-                    Datum::String(&source_id.to_string()),
+                    Datum::String(&source_id.source_id.to_string()),
                     Datum::String(&partition_id),
                     Datum::Int64(diff_vector[0]),
                     Datum::Int64(diff_vector[1]),

test/testdrive/kafka-stats.td

@@ -83,6 +83,18 @@ b  sum
 > SELECT count(*) FROM mz_kafka_consumer_statistics where consumer_lag = 0
 1
 
+# If we change the message encoding and/or the broker version, these results may change
+> SELECT partition_id, rx_msgs, rx_bytes, tx_msgs, tx_bytes, lo_offset, hi_offset, ls_offset, app_offset, consumer_lag FROM mz_kafka_consumer_statistics;
+partition_id  rx_msgs  rx_bytes  tx_msgs  tx_bytes  lo_offset  hi_offset  ls_offset  app_offset  consumer_lag
+-------------------------------------------------------------------------------------------------------------
+0  4  28  0  0  0  4  4  4 0
+
+# Verify that we can join against mz_source_info
+> SELECT mz_kafka_consumer_statistics.rx_msgs FROM mz_kafka_consumer_statistics INNER JOIN mz_source_info USING (source_id, partition_id) WHERE mz_source_info.source_name like 'kafka-%%';
+rx_msgs
+-------
+4
+
 # Drop the sources and verify that metrics have been removed
 > DROP VIEW test1