Flink: Allow setting slot sharing group for fine-grained resource management

[sqd]

Currently all operators created by the dynamic sink are part of the default slot sharing group, and thus getting an equal share of the resources on taskmanagers. However, it is usually the case that the sink and the generator operators are far more resource-heavy than the rest of the operators, making the default resource allocation inefficient.

Flink already supports fine-grained resource management mechanism to support use cases exactly like this. This change adds support to wire the dynamic sink into that system, by allowing the users to set slot sharing groups for 1. the shuffle writer 2. the generator+the forward writer -- they need to share the same slot sharing group to enable operator chaining.

[sqd]

@mxm @pvary Would appreciate if you could take a look

[sqd]

We have been running this internally for a while now. This allows Flink pipelines using the Iceberg dynamic sink to very flexibly slice the taskmanager resources like this:

[pvary]

CC: @mxm, @Guosmilesmile

[mxm]

How important is it for the slot sharing groups to be set explicitly? Could we add an option like disableSlotSharing() to put the components into different slots?

[sqd]

How important is it for the slot sharing groups to be set explicitly? Could we add an option like disableSlotSharing() to put the components into different slots?

The goal is to allow tailoring the resources allocated to each operator using Flink fine-grained resource management, so the user needs to pass in an SSG like this

sinkBuilder
  .generatorSlotSharingGroup(
    SlotSharingGroup.newBuilder('generator-ssg')
      .setCpuCores(1)
      .setTaskHeapMemoryMB(512)
      .build())
  .otherSinkBuilderOptions(...)
...

[sqd]

@mxm I also think it's a bit confusing that Flink uses "slot sharing group" which seems to imply some sort of resource isolation mechanism to manage resources, but here we are. :-)