pnowojski commented on a change in pull request #16713:
URL: https://github.com/apache/flink/pull/16713#discussion_r684027938
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -397,11 +398,24 @@ protected StreamTask(
this.throughputCalculator = environment.getThroughputMeter();
this.bufferDebloatPeriod =
getTaskConfiguration().get(BUFFER_DEBLOAT_PERIOD).toMillis();
- this.bufferDebloater =
-
getTaskConfiguration().get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED)
- ? new BufferDebloater(
- getTaskConfiguration(),
getEnvironment().getAllInputGates())
- : null;
+ if
(getTaskConfiguration().get(TaskManagerOptions.BUFFER_DEBLOAT_ENABLED)) {
+ this.bufferDebloater =
+ new BufferDebloater(
+ getTaskConfiguration(),
getEnvironment().getAllInputGates());
+ environment
+ .getMetricGroup()
+ .gauge(
+ MetricNames.ESTIMATED_TIME_TO_CONSUME_BUFFERS,
+ () ->
+ bufferDebloater
+
.getLastEstimatedTimeToConsumeBuffers()
+ .toMillis());
Review comment:
Unfortunately no. I would like it to work like that in the long term.
However as it is right now, metrics are probed from external threads and the
conscious decision is to have those access thread unsafe - decision was that
it's better to have some infrequent incorrect metric readings rather then
synchornize the code and adding huge overheads in runtime.
--
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]
