You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2022/03/03 12:31:19 UTC
[flink] branch release-1.13 updated: [FLINK-26407][end-to-end-tests] Increase timeouts for MetricsAvailabilityITCase.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.13 by this push:
new 0bfb320 [FLINK-26407][end-to-end-tests] Increase timeouts for MetricsAvailabilityITCase.
0bfb320 is described below
commit 0bfb320bc456397c211e1d2263a4199d2370a8d0
Author: David Moravek <dm...@apache.org>
AuthorDate: Tue Mar 1 17:04:51 2022 +0100
[FLINK-26407][end-to-end-tests] Increase timeouts for MetricsAvailabilityITCase.
---
.../metrics/tests/MetricsAvailabilityITCase.java | 40 ++++++++++++----------
1 file changed, 22 insertions(+), 18 deletions(-)
diff --git a/flink-end-to-end-tests/flink-metrics-availability-test/src/test/java/org/apache/flink/metrics/tests/MetricsAvailabilityITCase.java b/flink-end-to-end-tests/flink-metrics-availability-test/src/test/java/org/apache/flink/metrics/tests/MetricsAvailabilityITCase.java
index 95fd9e0..e5b711b 100644
--- a/flink-end-to-end-tests/flink-metrics-availability-test/src/test/java/org/apache/flink/metrics/tests/MetricsAvailabilityITCase.java
+++ b/flink-end-to-end-tests/flink-metrics-availability-test/src/test/java/org/apache/flink/metrics/tests/MetricsAvailabilityITCase.java
@@ -60,8 +60,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -92,24 +90,25 @@ public class MetricsAvailabilityITCase extends TestLogger {
@Test
public void testReporter() throws Exception {
+ final Deadline deadline = Deadline.fromNow(Duration.ofMinutes(10));
try (ClusterController ignored = dist.startCluster(1)) {
final RestClient restClient =
new RestClient(
RestClientConfiguration.fromConfiguration(new Configuration()),
scheduledExecutorService);
- checkJobManagerMetricAvailability(restClient);
+ checkJobManagerMetricAvailability(restClient, deadline);
- final Collection<ResourceID> taskManagerIds = getTaskManagerIds(restClient);
+ final Collection<ResourceID> taskManagerIds = getTaskManagerIds(restClient, deadline);
for (final ResourceID taskManagerId : taskManagerIds) {
- checkTaskManagerMetricAvailability(restClient, taskManagerId);
+ checkTaskManagerMetricAvailability(restClient, taskManagerId, deadline);
}
}
}
- private static void checkJobManagerMetricAvailability(final RestClient restClient)
- throws Exception {
+ private static void checkJobManagerMetricAvailability(
+ final RestClient restClient, final Deadline deadline) throws Exception {
final JobManagerMetricsHeaders headers = JobManagerMetricsHeaders.getInstance();
final JobManagerMetricsMessageParameters parameters =
headers.getUnresolvedMessageParameters();
@@ -120,11 +119,12 @@ public class MetricsAvailabilityITCase extends TestLogger {
() ->
restClient.sendRequest(
HOST, PORT, headers, parameters, EmptyRequestBody.getInstance()),
- getMetricNamePredicate("numRegisteredTaskManagers"));
+ getMetricNamePredicate("numRegisteredTaskManagers"),
+ deadline);
}
- private static Collection<ResourceID> getTaskManagerIds(final RestClient restClient)
- throws Exception {
+ private static Collection<ResourceID> getTaskManagerIds(
+ final RestClient restClient, final Deadline deadline) throws Exception {
final TaskManagersHeaders headers = TaskManagersHeaders.getInstance();
final TaskManagersInfo response =
@@ -136,7 +136,8 @@ public class MetricsAvailabilityITCase extends TestLogger {
headers,
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance()),
- taskManagersInfo -> !taskManagersInfo.getTaskManagerInfos().isEmpty());
+ taskManagersInfo -> !taskManagersInfo.getTaskManagerInfos().isEmpty(),
+ deadline);
return response.getTaskManagerInfos().stream()
.map(TaskManagerInfo::getResourceId)
@@ -144,7 +145,8 @@ public class MetricsAvailabilityITCase extends TestLogger {
}
private static void checkTaskManagerMetricAvailability(
- final RestClient restClient, final ResourceID taskManagerId) throws Exception {
+ final RestClient restClient, final ResourceID taskManagerId, final Deadline deadline)
+ throws Exception {
final TaskManagerMetricsHeaders headers = TaskManagerMetricsHeaders.getInstance();
final TaskManagerMetricsMessageParameters parameters =
headers.getUnresolvedMessageParameters();
@@ -156,13 +158,15 @@ public class MetricsAvailabilityITCase extends TestLogger {
() ->
restClient.sendRequest(
HOST, PORT, headers, parameters, EmptyRequestBody.getInstance()),
- getMetricNamePredicate("Status.Network.TotalMemorySegments"));
+ getMetricNamePredicate("Status.Network.TotalMemorySegments"),
+ deadline);
}
private static <X> X fetchMetric(
final SupplierWithException<CompletableFuture<X>, IOException> clientOperation,
- final Predicate<X> predicate)
- throws InterruptedException, ExecutionException, TimeoutException {
+ final Predicate<X> predicate,
+ final Deadline deadline)
+ throws InterruptedException, ExecutionException {
final CompletableFuture<X> responseFuture =
FutureUtils.retrySuccessfulWithDelay(
() -> {
@@ -172,12 +176,12 @@ public class MetricsAvailabilityITCase extends TestLogger {
throw new RuntimeException(e);
}
},
- Time.seconds(1),
- Deadline.fromNow(Duration.ofSeconds(5)),
+ Time.milliseconds(100),
+ deadline,
predicate,
new ScheduledExecutorServiceAdapter(scheduledExecutorService));
- return responseFuture.get(30, TimeUnit.SECONDS);
+ return responseFuture.get();
}
private static Predicate<MetricCollectionResponseBody> getMetricNamePredicate(