Prevent the potential for a panic in the string_to_timestamp coercions

The original code was turning nanos into micros anyways, so we can use
the more safe timestamp_micros() code.

For timestamp_nanos_opt() it can return None if the dates that cannot be
represented as nanoseconds between 1677-09-21T00:12:43.145224192 and
2262-04-11T23:47:16.854775807.

Signed-off-by: R. Tyler Croy <rtyler@brokenco.de>

src/coercions.rs

@@ -122,6 +122,9 @@ fn apply_coercion(value: &mut Value, node: &CoercionNode) {
     }
 }
 
+/// Convert a given datetime looking string into microseconds using chrono's [DateTime] parsing
+///
+/// Since this will convert to microseconds, if the value is outside of the realm of conversion a None is returned
 fn string_to_timestamp(string: &str) -> Option<Value> {
     let parsed = DateTime::from_str(string);
     if let Err(e) = parsed {
@@ -131,23 +134,25 @@ fn string_to_timestamp(string: &str) -> Option<Value> {
             e
         )
     }
-    parsed.ok().map(|dt: DateTime<Utc>| {
-        Value::Number(
-            (dt.timestamp_nanos_opt()
-                .expect("Failed to turn timestamp nanoseconds")
-                / 1000)
-                .into(),
-        )
-    })
+    parsed
+        .ok()
+        .map(|dt: DateTime<Utc>| Value::Number(dt.timestamp_micros().into()))
 }
 
 #[cfg(test)]
 mod tests {
-
     use super::*;
-    // use maplit::hashmap;
     use serde_json::json;
 
+    #[test]
+    fn test_string_to_timestamp() {
+        let result = string_to_timestamp("2010-01-01T22:11:58Z");
+        assert!(result.is_some());
+        // exceeds nanos size
+        let result = string_to_timestamp("2400-01-01T22:11:58Z");
+        assert!(result.is_some());
+    }
+
     lazy_static! {
         static ref SCHEMA: Value = json!({
             "type": "struct",

src/main.rs

@@ -476,8 +476,7 @@ fn convert_matches_to_message_format(
             .map(MessageFormat::Avro);
     }
 
-    return to_schema_source(ingest_matches.get_one::<String>("json"), true)
-        .map(MessageFormat::Json);
+    to_schema_source(ingest_matches.get_one::<String>("json"), true).map(MessageFormat::Json)
 }
 
 #[cfg(test)]