(
msg: &BorrowedMessage<'_>,
metadata_columns: &[KafkaMetadataKind],
)
| 1346 | } |
| 1347 | |
| 1348 | fn construct_source_message( |
| 1349 | msg: &BorrowedMessage<'_>, |
| 1350 | metadata_columns: &[KafkaMetadataKind], |
| 1351 | ) -> ( |
| 1352 | Result<SourceMessage, KafkaHeaderParseError>, |
| 1353 | (PartitionId, MzOffset), |
| 1354 | ) { |
| 1355 | let pid = msg.partition(); |
| 1356 | let Ok(offset) = u64::try_from(msg.offset()) else { |
| 1357 | panic!( |
| 1358 | "got negative offset ({}) from otherwise non-error'd kafka message", |
| 1359 | msg.offset() |
| 1360 | ); |
| 1361 | }; |
| 1362 | |
| 1363 | let mut metadata = Row::default(); |
| 1364 | let mut packer = metadata.packer(); |
| 1365 | for kind in metadata_columns { |
| 1366 | match kind { |
| 1367 | KafkaMetadataKind::Partition => packer.push(Datum::from(pid)), |
| 1368 | KafkaMetadataKind::Offset => packer.push(Datum::UInt64(offset)), |
| 1369 | KafkaMetadataKind::Timestamp => { |
| 1370 | let ts = msg |
| 1371 | .timestamp() |
| 1372 | .to_millis() |
| 1373 | .expect("kafka sources always have upstream_time"); |
| 1374 | |
| 1375 | let d: Datum = DateTime::from_timestamp_millis(ts) |
| 1376 | .and_then(|dt| { |
| 1377 | let ct: Option<CheckedTimestamp<NaiveDateTime>> = |
| 1378 | dt.naive_utc().try_into().ok(); |
| 1379 | ct |
| 1380 | }) |
| 1381 | .into(); |
| 1382 | packer.push(d) |
| 1383 | } |
| 1384 | KafkaMetadataKind::Header { key, use_bytes } => { |
| 1385 | match msg.headers() { |
| 1386 | Some(headers) => { |
| 1387 | let d = headers |
| 1388 | .iter() |
| 1389 | .filter(|header| header.key == key) |
| 1390 | .last() |
| 1391 | .map(|header| match header.value { |
| 1392 | Some(v) => { |
| 1393 | if *use_bytes { |
| 1394 | Ok(Datum::Bytes(v)) |
| 1395 | } else { |
| 1396 | match str::from_utf8(v) { |
| 1397 | Ok(str) => Ok(Datum::String(str)), |
| 1398 | Err(_) => Err(KafkaHeaderParseError::Utf8Error { |
| 1399 | key: key.clone(), |
| 1400 | raw: v.to_vec(), |
| 1401 | }), |
| 1402 | } |
| 1403 | } |
| 1404 | } |
| 1405 | None => Ok(Datum::Null), |
no test coverage detected