skywalker0618 commented on code in PR #20499:
URL: https://github.com/apache/hudi/pull/20499#discussion_r4474120838
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java:
##########
@@ -37,10 +41,24 @@
*
* <p>The breakdown of each sync, from the init duration to the schema evolved
count, is only
* registered when {@code
hoodie.datasource.hive_sync.detailed_metrics.enabled} is set.
+ *
+ * <p>Obtain an instance with {@link #register}. Flink keeps one coordinator
metric group per operator
+ * for the life of the job, but creates a new coordinator whenever it rebuilds
the execution graph,
+ * as the adaptive scheduler does on every restart and rescale. That group
ignores a metric registered
+ * under a name it already holds, so metrics owned by each coordinator would
stay frozen at the first
+ * coordinator's values. Every coordinator of the same operator in the same
job therefore shares one
+ * instance.
*/
public class FlinkHiveSyncMetrics extends HoodieFlinkMetrics {
private static final int HISTOGRAM_WINDOW_SIZE = 100;
+ /**
+ * The instance shared by the coordinators of each operator, keyed by job
and operator ID. Entries
+ * are never removed: a coordinator closing does not tell whether another
will replace it, and one
+ * small instance per Hudi write operator per job is all this holds.
+ */
+ private static final Map<String, FlinkHiveSyncMetrics> SHARED_METRICS = new
ConcurrentHashMap<>();
Review Comment:
Great suggestion, fixed!
--
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]
