public static TaskExecutorProcessSpec processSpecFromWorkerResourceSpec(
            final Configuration config, final WorkerResourceSpec workerResourceSpec) {

        final MemorySize frameworkHeapMemorySize =
                TaskExecutorFlinkMemoryUtils.getFrameworkHeapMemorySize(config);
        final MemorySize frameworkOffHeapMemorySize =
                TaskExecutorFlinkMemoryUtils.getFrameworkOffHeapMemorySize(config);

        final TaskExecutorFlinkMemory flinkMemory =
                new TaskExecutorFlinkMemory(
                        frameworkHeapMemorySize,
                        frameworkOffHeapMemorySize,
                        workerResourceSpec.getTaskHeapSize(),
                        workerResourceSpec.getTaskOffHeapSize(),
                        workerResourceSpec.getNetworkMemSize(),
                        workerResourceSpec.getManagedMemSize());

        final JvmMetaspaceAndOverhead jvmMetaspaceAndOverhead =
                PROCESS_MEMORY_UTILS.deriveJvmMetaspaceAndOverheadFromTotalFlinkMemory(
                        config, flinkMemory.getTotalFlinkMemorySize());

        return new TaskExecutorProcessSpec(
                workerResourceSpec.getCpuCores(),
                flinkMemory,
                jvmMetaspaceAndOverhead,
                workerResourceSpec.getNumSlots(),
                workerResourceSpec.getExtendedResources().values());
    }
@Test
    void testProcessSpecFromWorkerResourceSpec() {
        final WorkerResourceSpec workerResourceSpec =
                new WorkerResourceSpec.Builder()
                        .setCpuCores(1.0)
                        .setTaskHeapMemoryMB(100)
                        .setTaskOffHeapMemoryMB(200)
                        .setNetworkMemoryMB(300)
                        .setManagedMemoryMB(400)
                        .setNumSlots(5)
                        .setExtendedResource(new ExternalResource(EXTERNAL_RESOURCE_NAME_1, 1))
                        .build();
        final TaskExecutorProcessSpec taskExecutorProcessSpec =
                TaskExecutorProcessUtils.processSpecFromWorkerResourceSpec(
                        new Configuration(), workerResourceSpec);
        assertThat(taskExecutorProcessSpec.getCpuCores())
                .isEqualTo(workerResourceSpec.getCpuCores());
        assertThat(taskExecutorProcessSpec.getTaskHeapSize())
                .isEqualTo(workerResourceSpec.getTaskHeapSize());
        assertThat(taskExecutorProcessSpec.getTaskOffHeapSize())
                .isEqualTo(workerResourceSpec.getTaskOffHeapSize());
        assertThat(taskExecutorProcessSpec.getNetworkMemSize())
                .isEqualTo(workerResourceSpec.getNetworkMemSize());
        assertThat(taskExecutorProcessSpec.getManagedMemorySize())
                .isEqualTo(workerResourceSpec.getManagedMemSize());
        assertThat(taskExecutorProcessSpec.getNumSlots())
                .isEqualTo(workerResourceSpec.getNumSlots());
        assertThat(taskExecutorProcessSpec.getExtendedResources())
                .isEqualTo(workerResourceSpec.getExtendedResources());
    }