XComp commented on code in PR #27257:
URL: https://github.com/apache/flink/pull/27257#discussion_r2661472701
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java:
##########
@@ -1058,4 +1062,98 @@ void
testClearResourceRequirementsWithPendingTaskManager() throws Exception {
}
};
}
+
+ @Test
+ void testMetricsUpdate() throws Exception {
+ final AtomicReference<Gauge<Long>> slotsAvailableGauge = new
AtomicReference<>();
+ final AtomicReference<Gauge<Long>> slotsTotalGauge = new
AtomicReference<>();
+
+ final MetricRegistry metricRegistry =
+ TestingMetricRegistry.builder()
+ .setRegisterConsumer(
+ (metric, name, group) -> {
+ if
(name.equals(MetricNames.TASK_SLOTS_AVAILABLE)) {
+ slotsAvailableGauge.set((Gauge<Long>)
metric);
+ } else if
(name.equals(MetricNames.TASK_SLOTS_TOTAL)) {
+ slotsTotalGauge.set((Gauge<Long>)
metric);
+ }
+ })
+ .build();
+
+ final Context context = new Context();
+ context.setSlotManagerMetricGroup(
+ SlotManagerMetricGroup.create(metricRegistry, "localhost"));
+ final ManuallyTriggeredScheduledExecutor scheduledExecutor =
+ new ManuallyTriggeredScheduledExecutor();
+ context.setScheduledExecutor(scheduledExecutor);
+ final TaskExecutorConnection executorConnection1 =
createTaskExecutorConnection();
+ final TaskExecutorConnection executorConnection2 =
createTaskExecutorConnection();
+
+ context.runTest(
+ () -> {
+
assertThat(slotsAvailableGauge.get().getValue()).isEqualTo(0);
+ assertThat(slotsTotalGauge.get().getValue()).isEqualTo(0);
+
+ final CompletableFuture<SlotManager.RegistrationResult>
+ registerTaskManagerFuture1 = new
CompletableFuture<>();
+ context.runInMainThreadAndWait(
+ () ->
+ registerTaskManagerFuture1.complete(
+ context.getSlotManager()
+ .registerTaskManager(
+
executorConnection1,
+ new SlotReport(),
+
DEFAULT_TOTAL_RESOURCE_PROFILE,
+
DEFAULT_SLOT_RESOURCE_PROFILE)));
+
assertThat(assertFutureCompleteAndReturn(registerTaskManagerFuture1))
+ .isEqualTo(SlotManager.RegistrationResult.SUCCESS);
+
+ final CompletableFuture<SlotManager.RegistrationResult>
+ registerTaskManagerFuture2 = new
CompletableFuture<>();
+ context.runInMainThreadAndWait(
+ () ->
+ registerTaskManagerFuture2.complete(
+ context.getSlotManager()
+ .registerTaskManager(
+
executorConnection2,
+ new SlotReport(
+
createAllocatedSlotStatus(
+
new JobID(),
+
new AllocationID(),
+
DEFAULT_SLOT_RESOURCE_PROFILE)),
+
DEFAULT_TOTAL_RESOURCE_PROFILE,
+
DEFAULT_SLOT_RESOURCE_PROFILE)));
+
assertThat(assertFutureCompleteAndReturn(registerTaskManagerFuture2))
+ .isEqualTo(SlotManager.RegistrationResult.SUCCESS);
+
+ // triggers the metric update task on the main thread
+ scheduledExecutor.triggerPeriodicScheduledTasks();
+ context.runInMainThreadAndWait(() -> {});
+
+ assertThat(slotsTotalGauge.get().getValue())
+ .isEqualTo(2 * DEFAULT_NUM_SLOTS_PER_WORKER);
+ assertThat(slotsAvailableGauge.get().getValue())
+ .isEqualTo(2 * DEFAULT_NUM_SLOTS_PER_WORKER - 1);
+
+ final CompletableFuture<Boolean>
unRegisterTaskManagerFuture =
+ new CompletableFuture<>();
+ context.runInMainThreadAndWait(
+ () ->
+ unRegisterTaskManagerFuture.complete(
+ context.getSlotManager()
+ .unregisterTaskManager(
+
executorConnection2.getInstanceID(),
+ TEST_EXCEPTION)));
+
assertThat(assertFutureCompleteAndReturn(unRegisterTaskManagerFuture)).isTrue();
+
+ // triggers the metric update task on the main thread
Review Comment:
```suggestion
// triggers the metric update task on the main thread
and waits for the main thread to process queued up callbacks
```
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManagerTest.java:
##########
@@ -1058,4 +1062,98 @@ void
testClearResourceRequirementsWithPendingTaskManager() throws Exception {
}
};
}
+
+ @Test
+ void testMetricsUpdate() throws Exception {
+ final AtomicReference<Gauge<Long>> slotsAvailableGauge = new
AtomicReference<>();
+ final AtomicReference<Gauge<Long>> slotsTotalGauge = new
AtomicReference<>();
+
+ final MetricRegistry metricRegistry =
+ TestingMetricRegistry.builder()
+ .setRegisterConsumer(
+ (metric, name, group) -> {
+ if
(name.equals(MetricNames.TASK_SLOTS_AVAILABLE)) {
+ slotsAvailableGauge.set((Gauge<Long>)
metric);
+ } else if
(name.equals(MetricNames.TASK_SLOTS_TOTAL)) {
+ slotsTotalGauge.set((Gauge<Long>)
metric);
+ }
+ })
+ .build();
+
+ final Context context = new Context();
+ context.setSlotManagerMetricGroup(
+ SlotManagerMetricGroup.create(metricRegistry, "localhost"));
+ final ManuallyTriggeredScheduledExecutor scheduledExecutor =
+ new ManuallyTriggeredScheduledExecutor();
+ context.setScheduledExecutor(scheduledExecutor);
+ final TaskExecutorConnection executorConnection1 =
createTaskExecutorConnection();
+ final TaskExecutorConnection executorConnection2 =
createTaskExecutorConnection();
+
+ context.runTest(
+ () -> {
+
assertThat(slotsAvailableGauge.get().getValue()).isEqualTo(0);
+ assertThat(slotsTotalGauge.get().getValue()).isEqualTo(0);
+
+ final CompletableFuture<SlotManager.RegistrationResult>
+ registerTaskManagerFuture1 = new
CompletableFuture<>();
+ context.runInMainThreadAndWait(
+ () ->
+ registerTaskManagerFuture1.complete(
+ context.getSlotManager()
+ .registerTaskManager(
+
executorConnection1,
+ new SlotReport(),
+
DEFAULT_TOTAL_RESOURCE_PROFILE,
+
DEFAULT_SLOT_RESOURCE_PROFILE)));
+
assertThat(assertFutureCompleteAndReturn(registerTaskManagerFuture1))
+ .isEqualTo(SlotManager.RegistrationResult.SUCCESS);
+
+ final CompletableFuture<SlotManager.RegistrationResult>
+ registerTaskManagerFuture2 = new
CompletableFuture<>();
+ context.runInMainThreadAndWait(
+ () ->
+ registerTaskManagerFuture2.complete(
+ context.getSlotManager()
+ .registerTaskManager(
+
executorConnection2,
+ new SlotReport(
+
createAllocatedSlotStatus(
+
new JobID(),
+
new AllocationID(),
+
DEFAULT_SLOT_RESOURCE_PROFILE)),
+
DEFAULT_TOTAL_RESOURCE_PROFILE,
+
DEFAULT_SLOT_RESOURCE_PROFILE)));
+
assertThat(assertFutureCompleteAndReturn(registerTaskManagerFuture2))
+ .isEqualTo(SlotManager.RegistrationResult.SUCCESS);
+
+ // triggers the metric update task on the main thread
Review Comment:
```suggestion
// triggers the metric update task on the main thread
and waits for the main thread to process queued up callbacks
```
--
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]
