@PublicEvolving
    public StreamExecutionEnvironment registerSlotSharingGroup(SlotSharingGroup slotSharingGroup) {
        final ResourceSpec resourceSpec =
                SlotSharingGroupUtils.extractResourceSpec(slotSharingGroup);
        if (!resourceSpec.equals(ResourceSpec.UNKNOWN)) {
            this.slotSharingGroupResources.put(
                    slotSharingGroup.getName(),
                    ResourceProfile.fromResourceSpec(
                            SlotSharingGroupUtils.extractResourceSpec(slotSharingGroup),
                            MemorySize.ZERO));
        }
        return this;
    }
@Test
    void testRegisterSlotSharingGroup() {
        final SlotSharingGroup ssg1 =
                SlotSharingGroup.newBuilder("ssg1").setCpuCores(1).setTaskHeapMemoryMB(100).build();
        final SlotSharingGroup ssg2 =
                SlotSharingGroup.newBuilder("ssg2").setCpuCores(2).setTaskHeapMemoryMB(200).build();
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.registerSlotSharingGroup(ssg1);
        env.registerSlotSharingGroup(ssg2);
        env.registerSlotSharingGroup(SlotSharingGroup.newBuilder("ssg3").build());

        final DataStream<Integer> source = env.fromData(1).slotSharingGroup("ssg1");
        source.map(value -> value).slotSharingGroup(ssg2).sinkTo(new DiscardingSink<>());

        final StreamGraph streamGraph = env.getStreamGraph();
        assertThat(streamGraph.getSlotSharingGroupResource("ssg1").get())
                .isEqualTo(ResourceProfile.fromResources(1, 100));
        assertThat(streamGraph.getSlotSharingGroupResource("ssg2").get())
                .isEqualTo(ResourceProfile.fromResources(2, 200));
        assertThat(streamGraph.getSlotSharingGroupResource("ssg3")).isNotPresent();
    }