You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Yang-LI-CS (via GitHub)" <gi...@apache.org> on 2024/01/02 21:02:47 UTC
[PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Yang-LI-CS opened a new pull request, #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743
<!--
*Thank you very much for contributing to the Apache Flink Kubernetes Operator - we are happy that you want to help us improve the project. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix][docs] Fix typo in event time introduction` or `[hotfix][javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can read more on how we use GitHub Actions for CI [here](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/development/guide/#cicd).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Fix the getNumRecordsInPerSecond Utility Function to addresses the redundant check and ensures correct metric fetching for non-source operators.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changes to the `CustomResourceDescriptors`: no
- Core observer or reconciler logic that is regularly executed: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440322610
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,21 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // Generate numRecordsInPerSecond from 3 metrics:
+ // 1. If the vertex is not the source, use NUM_RECORDS_IN_PER_SEC metric
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // 2. If the vertex is the source, use SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric first.
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
+ // 3. If the vertex is the source and SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric not available
+ // then use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC metric
Review Comment:
```suggestion
// 3. If the vertex contains a source operator which does not emit input metrics, use output metrics instead.
// then use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC metric
```
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,21 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // Generate numRecordsInPerSecond from 3 metrics:
+ // 1. If the vertex is not the source, use NUM_RECORDS_IN_PER_SEC metric
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // 2. If the vertex is the source, use SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric first.
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
Review Comment:
```suggestion
// 2. If the former is unavailable and the vertex contains a source operator, use the corresponding source operator metric.
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
```
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,21 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // Generate numRecordsInPerSecond from 3 metrics:
+ // 1. If the vertex is not the source, use NUM_RECORDS_IN_PER_SEC metric
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
Review Comment:
```suggestion
// 1. If available, directly use the NUM_RECORDS_IN_PER_SEC task metric.
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1441197747
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,23 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // Generate numRecordsInPerSecond from 3 metrics:
+ // 1. If available, directly use the NUM_RECORDS_IN_PER_SEC task metric.
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // 2. If the former is unavailable and the vertex contains a source operator, use the
+ // corresponding source operator metric.
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
+ // 3. If the vertex contains a source operator which does not emit input metrics, use output
+ // metrics instead.
+ // then use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC metric
Review Comment:
```suggestion
```
Is this leftover?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#issuecomment-1875354667
> Looks good! I think this logic needs some comments because it is hard to understand the reasoning behind the logic. I added some additional suggestions.
Thanks @mxm I have added your suggestion :)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440186827
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -221,7 +221,7 @@ private static double getNumRecordsInPerSecond(
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
- if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
+ if (!isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
Review Comment:
I have just add some comments to clarify this at last 🙏
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440178966
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -221,7 +221,7 @@ private static double getNumRecordsInPerSecond(
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
- if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
+ if (!isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
Review Comment:
@1996fanrui thanks, I'll add some comments and do as your suggestion 👍
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1441489844
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,23 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // Generate numRecordsInPerSecond from 3 metrics:
+ // 1. If available, directly use the NUM_RECORDS_IN_PER_SEC task metric.
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // 2. If the former is unavailable and the vertex contains a source operator, use the
+ // corresponding source operator metric.
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
+ // 3. If the vertex contains a source operator which does not emit input metrics, use output
+ // metrics instead.
+ // then use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC metric
Review Comment:
@1996fanrui indeed, I'll remove it
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440197015
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,20 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // If the vertex is not the source, use NUM_RECORDS_IN_PER_SEC metric
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // If the vertex is the source, use SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
+ // If the vertex is the source and SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric not available
+ // then use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC metric
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
Review Comment:
```suggestion
// Generate numRecordsInPerSecond from 3 metrics:
// 1. If the vertex is not the source, use NUM_RECORDS_IN_PER_SEC metric
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
// 2. If the vertex is the source, use SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric first.
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
// 3. If the vertex is the source and SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric not available
// then use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC metric
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
```
How about adding the `Generate numRecordsInPerSecond from 3 metrics:` and the `1. 2. 3.`?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui merged PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440202523
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,20 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // If the vertex is not the source, use NUM_RECORDS_IN_PER_SEC metric
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // If the vertex is the source, use SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
+ // If the vertex is the source and SOURCE_TASK_NUM_RECORDS_IN_PER_SEC metric not available
+ // then use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC metric
if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
Review Comment:
Good idea! 🙏
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440519499
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,21 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // Generate numRecordsInPerSecond from 3 metrics:
+ // 1. If available, directly use the NUM_RECORDS_IN_PER_SEC task metric.
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // 2. If the former is unavailable and the vertex contains a source operator, use the corresponding source operator metric.
Review Comment:
We might have to run `mvn spotless:apply`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440104568
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -221,7 +221,7 @@ private static double getNumRecordsInPerSecond(
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
- if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
+ if (!isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
Review Comment:
Thanks @Yang-LI-CS for the PR!
These 2 conditions are indeed the same, it maybe unexpected intuitively. Maybe here logic is numRecordsInPerSecond from 3 metrics:
1. FlinkMetric.NUM_RECORDS_IN_PER_SEC
2. FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC
3. FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC
When 1 is unavailable, get it from 2. When 2 is unavailable, get it from 3.
I'm not sure about it. Need to confirm with @mxm .
- If my guess is right, it's not a bug. And I suggest this PR add a little comments to clarify it.
- If my guess is wrong, it's a bug. We can fix it.
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440183532
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -221,7 +221,7 @@ private static double getNumRecordsInPerSecond(
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
- if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
+ if (!isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
Review Comment:
@1996fanrui This time I have just removed the unused if condition and add 1 comment.
To be sure, let's wait feedback from @mxm
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440186563
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -217,14 +217,13 @@ private static double getNumRecordsInPerSecond(
JobVertexID jobVertexID,
boolean isSource) {
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
- if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
- numRecordsInPerSecond =
- flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
- }
Review Comment:
@1996fanrui Sry, I have put it back and just add the comments
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "Yang-LI-CS (via GitHub)" <gi...@apache.org>.
Yang-LI-CS commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440549106
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -216,15 +216,21 @@ private static double getNumRecordsInPerSecond(
Map<FlinkMetric, AggregatedMetric> flinkMetrics,
JobVertexID jobVertexID,
boolean isSource) {
+ // Generate numRecordsInPerSecond from 3 metrics:
+ // 1. If available, directly use the NUM_RECORDS_IN_PER_SEC task metric.
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+ // 2. If the former is unavailable and the vertex contains a source operator, use the corresponding source operator metric.
Review Comment:
@mxm 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440105061
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -221,7 +221,7 @@ private static double getNumRecordsInPerSecond(
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
}
- if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
+ if (!isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
numRecordsInPerSecond =
flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC);
}
Review Comment:
By the way, current change doesn't make sense.
When a task isn't source, it doesn't have `SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC` metric.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33966] Fix the getNumRecordsInPerSecond Utility Function [flink-kubernetes-operator]
Posted by "1996fanrui (via GitHub)" <gi...@apache.org>.
1996fanrui commented on code in PR #743:
URL: https://github.com/apache/flink-kubernetes-operator/pull/743#discussion_r1440182777
##########
flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java:
##########
@@ -217,14 +217,13 @@ private static double getNumRecordsInPerSecond(
JobVertexID jobVertexID,
boolean isSource) {
var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
- if (isSource && (numRecordsInPerSecond == null || numRecordsInPerSecond.getSum() == 0)) {
- numRecordsInPerSecond =
- flinkMetrics.get(FlinkMetric.SOURCE_TASK_NUM_RECORDS_IN_PER_SEC);
- }
Review Comment:
Why the `SOURCE_TASK_NUM_RECORDS_IN_PER_SEC` part is removed?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org