Default to capturing Kafka consumer metrics once per second

src/sql/src/kafka_util.rs

@@ -13,9 +13,11 @@ use std::collections::BTreeMap;
 use std::convert;
 use std::fs::File;
 use std::io::Read;
+use std::option::Option;
 use std::sync::Mutex;
 
 use anyhow::bail;
+use chrono;
 use log::{debug, error, info, warn};
 use rdkafka::client::ClientContext;
 use rdkafka::consumer::{BaseConsumer, Consumer, ConsumerContext};
@@ -36,11 +38,12 @@ enum ValType {
 
 // Describes Kafka cluster configurations users can suppply using `CREATE
 // SOURCE...WITH (option_list)`.
-// TODO(sploiselle): Support overriding keys, default values.
+// TODO(sploiselle): Support overriding keys.
 struct Config {
     name: &'static str,
     val_type: ValType,
     transform: fn(String) -> String,
+    default: fn() -> Option<String>,
 }
 
 impl Config {
@@ -49,6 +52,7 @@ impl Config {
             name,
             val_type,
             transform: convert::identity,
+            default: || -> Option<String> { None },
         }
     }
 
@@ -69,6 +73,12 @@ impl Config {
         self
     }
 
+    // Allows for returning a default value for this configuration option
+    fn default(mut self, f: fn() -> Option<String>) -> Self {
+        self.default = f;
+        self
+    }
+
     // Get the appropriate String to use as the Kafka config key.
     fn get_key(&self) -> String {
         self.name.replace("_", ".")
@@ -105,7 +115,12 @@ fn extract(
                 Ok(v) => v,
                 Err(e) => bail!("Invalid WITH option {}={}: {}", config.name, v, e),
             },
-            None => continue,
+            None => match (config.default)() {
+                Some(v) => v,
+                None => {
+                    continue;
+                }
+            },
         };
         out.insert(config.get_key(), value);
     }
@@ -134,7 +149,8 @@ pub fn extract_config(
                 // The range of values comes from `statistics.interval.ms` in
                 // https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
                 ValType::Number(0, 86_400_000),
-            ),
+            )
+            .default(|| Some(chrono::Duration::seconds(1).num_milliseconds().to_string())),
             Config::new(
                 "topic_metadata_refresh_interval_ms",
                 // The range of values comes from `topic.metadata.refresh.interval.ms` in

test/testdrive/kafka-stats.td

@@ -22,7 +22,6 @@ $ kafka-create-topic topic=data
 
 > CREATE SOURCE data
   FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC 'testdrive-data-${testdrive.seed}'
-  WITH (statistics_interval_ms = 1000)
   FORMAT AVRO USING SCHEMA '${schema}'
 
 > CREATE MATERIALIZED VIEW test1 AS