ryerraguntla commented on code in PR #3140:
URL: https://github.com/apache/iggy/pull/3140#discussion_r3232893923
##########
core/connectors/sinks/influxdb_sink/src/lib.rs:
##########
@@ -479,20 +573,34 @@ impl InfluxDbSink {
message.timestamp
};
let ts = self.to_precision_timestamp(base_micros);
+ write!(buf, " {ts}").expect("infallible");
- {
- use std::fmt::Write as _;
- write!(buf, " {ts}").expect("write to String is infallible");
- }
-
- debug!(
- "InfluxDB sink ID: {} point — offset={}, raw_ts={},
influx_ts={ts}",
- self.id, message.offset, message.timestamp
- );
-
+ debug!("sink ID: {} — offset={}, ts={ts}", self.id, message.offset);
Ok(())
}
+ /// Build the newline-separated line-protocol body for a batch of messages.
+ /// Pure function — no I/O; extracted for testability. The empty-slice
path is
+ /// unreachable in production (process_batch returns early when messages
is empty)
+ /// but is exercised by unit tests for defensive completeness.
+ fn build_body(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ messages: &[ConsumedMessage],
+ ) -> Result<String, Error> {
+ // 1 KiB per message is a conservative estimate that accommodates JSON
+ // payloads without excessive reallocation.
+ let mut body = String::with_capacity(messages.len() * 1024);
Review Comment:
Done !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
