TanYuxin-tyx commented on code in PR #21132:
URL: https://github.com/apache/flink/pull/21132#discussion_r1004059827
##########
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricFetcherImpl.java:
##########
@@ -148,102 +156,106 @@ private void fetchMetrics() {
},
executor);
- CompletableFuture<Collection<String>>
queryServiceAddressesFuture =
-
leaderGateway.requestMetricQueryServiceAddresses(timeout);
-
- queryServiceAddressesFuture.whenCompleteAsync(
- (Collection<String> queryServiceAddresses, Throwable
throwable) -> {
+ List<CompletableFuture<Void>> waitingMetricsFutures = new
ArrayList<>();
+ CompletableFuture<Void> jmMetricsFuture =
queryJmMetricsFuture(leaderGateway);
+ waitingMetricsFutures.add(jmMetricsFuture);
+ jmMetricsFuture.whenCompleteAsync(
+ (ignore, throwable) -> {
if (throwable != null) {
- LOG.debug("Requesting paths for query services
failed.", throwable);
- } else {
- for (String queryServiceAddress :
queryServiceAddresses) {
-
retrieveAndQueryMetrics(queryServiceAddress);
- }
+ LOG.debug("Failed to fetch the leader's
metrics.", throwable);
}
- },
- executor);
-
- // TODO: Once the old code has been ditched, remove the
explicit TaskManager query
- // service discovery
- // TODO: and return it as part of
requestMetricQueryServiceAddresses. Moreover,
- // change the MetricStore such that
- // TODO: we don't have to explicitly retain the valid
TaskManagers, e.g. letting it
- // be a cache with expiry time
- CompletableFuture<Collection<Tuple2<ResourceID, String>>>
- taskManagerQueryServiceGatewaysFuture =
-
leaderGateway.requestTaskManagerMetricQueryServiceAddresses(
- timeout);
-
- taskManagerQueryServiceGatewaysFuture.whenCompleteAsync(
- (Collection<Tuple2<ResourceID, String>>
queryServiceGateways,
- Throwable throwable) -> {
+ });
+ CompletableFuture<Void> tmMetricsFuture =
queryTmMetricsFuture(leaderGateway);
+ waitingMetricsFutures.add(tmMetricsFuture);
+ tmMetricsFuture.whenCompleteAsync(
+ (ignore, throwable) -> {
if (throwable != null) {
- LOG.debug(
- "Requesting TaskManager's path for
query services failed.",
- throwable);
- } else {
- List<String> taskManagersToRetain =
- queryServiceGateways.stream()
- .map(
- (Tuple2<ResourceID,
String> tuple) -> {
-
queryServiceRetriever
-
.retrieveService(tuple.f1)
-
.thenAcceptAsync(
-
this::queryMetrics,
-
executor);
- return
tuple.f0.getResourceIdString();
- })
- .collect(Collectors.toList());
-
-
metrics.retainTaskManagers(taskManagersToRetain);
+ LOG.debug("Failed to fetch the TaskManager's
metrics.", throwable);
Review Comment:
ok, fixed.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
