pnowojski commented on code in PR #23908:
URL: https://github.com/apache/flink/pull/23908#discussion_r1441707237
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointStatsTracker.java:
##########
@@ -343,6 +376,62 @@ public void reportIncompleteStats(
}
}
+ public void reportInitializationStartTs(long initializationStartTs) {
+ jobInitializationMetricsBuilder =
+ Optional.of(
+ new JobInitializationMetricsBuilder(
+ totalNumberOfSubTasks, initializationStartTs));
+ }
+
+ public void reportInitializationMetrics(SubTaskInitializationMetrics
initializationMetrics) {
+ statsReadWriteLock.lock();
+ try {
+ if (!jobInitializationMetricsBuilder.isPresent()) {
+ LOG.warn(
+ "Attempted to report SubTaskInitializationMetrics [{}]
without jobInitializationMetricsBuilder present",
+ initializationMetrics);
+ return;
+ }
+ jobInitializationMetricsBuilder
+ .get()
+ .reportInitializationMetrics(initializationMetrics);
+ if (jobInitializationMetricsBuilder.get().isComplete()) {
+
traceInitializationMetrics(jobInitializationMetricsBuilder.get().build());
+ }
+ } catch (Exception ex) {
+ LOG.warn("Fail to log SubTaskInitializationMetrics[{}]", ex,
initializationMetrics);
Review Comment:
I don't think it's worth optimising this, as that's not an expected
exception, but a bug in the code.
Also currently there is an invariant `jobInitializationMetricsBuilder` is
not empty while metric could be collected. I'm not sure if change that you
propose would work out of the box, or if some `checkState` or sth else would
fail. So I would prefer to leave it as it is to save up a bit of work/time.
--
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]
