XComp commented on code in PR #23737:
URL: https://github.com/apache/flink/pull/23737#discussion_r1397153112
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java:
##########
@@ -217,43 +215,41 @@ private void assertSinkMetrics(
assertThat(subtaskWithMetrics, equalTo(numSplits));
}
- private void assertSinkCommitterMetrics(
- JobID jobId, int parallelism, Map<String, Long> expected) {
+ private void assertSinkCommitterMetrics(JobID jobId, Map<String, Long>
expected) {
List<OperatorMetricGroup> groups =
reporter.findOperatorMetricGroups(
jobId, TEST_SINK_NAME + ": " + DEFAULT_COMMITTER_NAME);
- assertThat(groups, hasSize(parallelism));
Map<String, Long> aggregated = new HashMap<>(6);
for (OperatorMetricGroup group : groups) {
Map<String, Metric> metrics = reporter.getMetricsByGroup(group);
- aggregated.merge(
- MetricNames.SUCCESSFUL_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.SUCCESSFUL_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.ALREADY_COMMITTED_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.ALREADY_COMMITTED_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.RETRIED_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.RETRIED_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.FAILED_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.FAILED_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.TOTAL_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.TOTAL_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.PENDING_COMMITTABLES,
- ((Gauge<Integer>)
metrics.get(MetricNames.PENDING_COMMITTABLES))
- .getValue()
- .longValue(),
- Long::sum);
+ if (metrics == null) {
+ continue;
+ }
Review Comment:
```suggestion
```
That doesn't look like it's necessary. `InMemoryReporter#getMetricsByGroup`
is always returning a `HashMap`. Or am I missing something here? :thinking:
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java:
##########
@@ -217,43 +215,41 @@ private void assertSinkMetrics(
assertThat(subtaskWithMetrics, equalTo(numSplits));
}
- private void assertSinkCommitterMetrics(
- JobID jobId, int parallelism, Map<String, Long> expected) {
+ private void assertSinkCommitterMetrics(JobID jobId, Map<String, Long>
expected) {
List<OperatorMetricGroup> groups =
reporter.findOperatorMetricGroups(
jobId, TEST_SINK_NAME + ": " + DEFAULT_COMMITTER_NAME);
- assertThat(groups, hasSize(parallelism));
Review Comment:
why is this assert not necessary anymore? (disclaimer: I'm not that familiar
with this part of the code :innocent: )
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java:
##########
@@ -217,43 +215,41 @@ private void assertSinkMetrics(
assertThat(subtaskWithMetrics, equalTo(numSplits));
}
- private void assertSinkCommitterMetrics(
- JobID jobId, int parallelism, Map<String, Long> expected) {
+ private void assertSinkCommitterMetrics(JobID jobId, Map<String, Long>
expected) {
List<OperatorMetricGroup> groups =
reporter.findOperatorMetricGroups(
jobId, TEST_SINK_NAME + ": " + DEFAULT_COMMITTER_NAME);
- assertThat(groups, hasSize(parallelism));
Map<String, Long> aggregated = new HashMap<>(6);
for (OperatorMetricGroup group : groups) {
Map<String, Metric> metrics = reporter.getMetricsByGroup(group);
- aggregated.merge(
- MetricNames.SUCCESSFUL_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.SUCCESSFUL_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.ALREADY_COMMITTED_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.ALREADY_COMMITTED_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.RETRIED_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.RETRIED_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.FAILED_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.FAILED_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.TOTAL_COMMITTABLES,
- ((Counter)
metrics.get(MetricNames.TOTAL_COMMITTABLES)).getCount(),
- Long::sum);
- aggregated.merge(
- MetricNames.PENDING_COMMITTABLES,
- ((Gauge<Integer>)
metrics.get(MetricNames.PENDING_COMMITTABLES))
- .getValue()
- .longValue(),
- Long::sum);
+ if (metrics == null) {
+ continue;
+ }
+
+ Gauge<Integer> pendingMetrics =
+ (Gauge<Integer>)
metrics.get(MetricNames.PENDING_COMMITTABLES);
Review Comment:
nit: could we move this closer to where it's actually used (i.e. before the
if condition for `pendingMetrics`) to improve code readability.
--
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-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
