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:30:36 UTC
[flink] branch release-1.14 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.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new 2b32f3c1 [FLINK-26407][end-to-end-tests] Increase timeouts for MetricsAvailabilityITCase.
2b32f3c1 is described below
commit 2b32f3c1b7227a577ae51fa007c816b093507c61
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 cf42948..0a584bd 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
@@ -59,8 +59,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;
@@ -91,22 +89,23 @@ 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(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();
@@ -117,11 +116,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 =
@@ -133,7 +133,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)
@@ -141,7 +142,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();
@@ -153,13 +155,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(
() -> {
@@ -169,12 +173,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(