public static void instantiateFlinkMemoryMetricGroup(
            MetricGroup parentMetricGroup,
            TaskSlotTable<?> taskSlotTable,
            Supplier<Long> managedMemoryTotalSupplier) {
        checkNotNull(parentMetricGroup);
        checkNotNull(taskSlotTable);
        checkNotNull(managedMemoryTotalSupplier);

        MetricGroup flinkMemoryMetricGroup =
                parentMetricGroup.addGroup(METRIC_GROUP_FLINK).addGroup(METRIC_GROUP_MEMORY);

        instantiateManagedMemoryMetrics(
                flinkMemoryMetricGroup, taskSlotTable, managedMemoryTotalSupplier);
    }
@Test
    void testManagedMemoryMetricsInitialization() throws MemoryAllocationException, FlinkException {
        final int maxMemorySize = 16284;
        final int numberOfAllocatedPages = 2;
        final int pageSize = 4096;
        final Object owner = new Object();

        final MemoryManager memoryManager = MemoryManager.create(maxMemorySize, pageSize);
        memoryManager.allocatePages(owner, numberOfAllocatedPages);
        final TaskManagerServices taskManagerServices =
                new TaskManagerServicesBuilder()
                        .setTaskSlotTable(
                                new TestingTaskSlotTable.TestingTaskSlotTableBuilder<Task>()
                                        .memoryManagerGetterReturns(memoryManager)
                                        .allActiveSlotAllocationIds(
                                                () -> Sets.newHashSet(new AllocationID()))
                                        .build())
                        .setManagedMemorySize(maxMemorySize)
                        .build();
        try {

            List<String> actualSubGroupPath = new ArrayList<>();
            final InterceptingOperatorMetricGroup metricGroup =
                    new InterceptingOperatorMetricGroup() {
                        @Override
                        public MetricGroup addGroup(String name) {
                            actualSubGroupPath.add(name);
                            return this;
                        }
                    };
            MetricUtils.instantiateFlinkMemoryMetricGroup(
                    metricGroup,
                    taskManagerServices.getTaskSlotTable(),
                    taskManagerServices::getManagedMemorySize);

            Gauge<Number> usedMetric = (Gauge<Number>) metricGroup.get("Used");
            Gauge<Number> maxMetric = (Gauge<Number>) metricGroup.get("Total");

            assertThat(usedMetric.getValue().intValue())
                    .isEqualTo(numberOfAllocatedPages * pageSize);
            assertThat(maxMetric.getValue().intValue()).isEqualTo(maxMemorySize);

            assertThat(actualSubGroupPath)
                    .containsAnyElementsOf(
                            Arrays.asList(
                                    METRIC_GROUP_FLINK,
                                    METRIC_GROUP_MEMORY,
                                    METRIC_GROUP_MANAGED_MEMORY));
        } finally {
            taskManagerServices.shutDown();
        }
    }