afedulov commented on code in PR #19660:
URL: https://github.com/apache/flink/pull/19660#discussion_r874800450
##########
flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/SourceMetricsITCase.java:
##########
@@ -169,45 +168,47 @@ private void assertSourceMetrics(
boolean hasTimestamps) {
List<OperatorMetricGroup> groups =
reporter.findOperatorMetricGroups(jobId,
"MetricTestingSource");
- assertThat(groups, hasSize(parallelism));
+ assertThat(groups).hasSize(parallelism);
int subtaskWithMetrics = 0;
for (OperatorMetricGroup group : groups) {
Map<String, Metric> metrics = reporter.getMetricsByGroup(group);
// there are only 2 splits assigned; so two groups will not update
metrics
if (group.getIOMetricGroup().getNumRecordsInCounter().getCount()
== 0) {
// assert that optional metrics are not initialized when no
split assigned
- assertThat(
- metrics.get(MetricNames.CURRENT_EMIT_EVENT_TIME_LAG),
-
isGauge(equalTo(InternalSourceReaderMetricGroup.UNDEFINED)));
- assertThat(metrics.get(MetricNames.WATERMARK_LAG),
nullValue());
+
assertThat(metrics.get(MetricNames.CURRENT_EMIT_EVENT_TIME_LAG))
+ .satisfies(
+ matching(
+ isGauge(
+ equalTo(
+
InternalSourceReaderMetricGroup
+ .UNDEFINED))));
+ assertThat(metrics.get(MetricNames.WATERMARK_LAG)).isNull();
continue;
}
subtaskWithMetrics++;
// I/O metrics
- assertThat(
- group.getIOMetricGroup().getNumRecordsInCounter(),
- isCounter(equalTo(processedRecordsPerSubtask)));
- assertThat(
- group.getIOMetricGroup().getNumBytesInCounter(),
- isCounter(
- equalTo(
- processedRecordsPerSubtask
- *
MockRecordEmitter.RECORD_SIZE_IN_BYTES)));
+ assertThat(group.getIOMetricGroup().getNumRecordsInCounter())
+
.satisfies(matching(isCounter(equalTo(processedRecordsPerSubtask))));
+ assertThat(group.getIOMetricGroup().getNumBytesInCounter())
+ .satisfies(
+ matching(
+ isCounter(
+ equalTo(
+ processedRecordsPerSubtask
+ * MockRecordEmitter
+
.RECORD_SIZE_IN_BYTES))));
// MockRecordEmitter is just incrementing errors every even record
- assertThat(
- metrics.get(MetricNames.NUM_RECORDS_IN_ERRORS),
- isCounter(equalTo(processedRecordsPerSubtask / 2)));
+ assertThat(metrics.get(MetricNames.NUM_RECORDS_IN_ERRORS))
+
.satisfies(matching(isCounter(equalTo(processedRecordsPerSubtask / 2))));
Review Comment:
Perfect, that is much better :+1:
--
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
