Cq-study commented on code in PR #681:
URL:
https://github.com/apache/doris-flink-connector/pull/681#discussion_r3673061507
##########
flink-doris-connector/flink-doris-connector-base/src/main/java/org/apache/doris/flink/sink/writer/DorisWriteMetrics.java:
##########
@@ -187,8 +192,10 @@ public void register(SinkWriterMetricGroup
sinkMetricGroup) {
private void flushSuccessLoad(RespContent responseContent) {
Optional.ofNullable(responseContent.getLoadBytes()).ifPresent(totalFlushLoadBytes::inc);
+
Optional.ofNullable(responseContent.getLoadBytes()).ifPresent(numBytesSend::inc);
Optional.ofNullable(responseContent.getNumberLoadedRows())
.ifPresent(totalFlushLoadedRows::inc);
+
Optional.ofNullable(responseContent.getNumberLoadedRows()).ifPresent(numRecordsSend::inc);
Review Comment:
Yes, you're right — per the `SinkWriterMetricGroup` javadoc,
`numRecordsSend` should count all records the writer has sent to the external
system, including those the downstream fails to persist, so incrementing with
`NumberLoadedRows` only on successful loads is not accurate.
I plan to move the counting to the write path instead: increment
`numRecordsSend` per record and `numBytesSend` by the payload size in
`DorisWriter#writeOneDorisRecord` when the record is written to the stream
load. This matches the "sent from Flink to Doris" semantics and also updates
continuously instead of only at checkpoint flush. The existing
connector-specific counters (`totalFlushLoadedRows`, etc.) would keep reporting
what Doris actually loaded. WDYT?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
