Reduce memory footprint on workload with thousand active partitions

create table source (
    `id` int,
    `data` string
) with (
    'connector' = 'datagen',
    'rows-per-second' = '100',
    'fields.id.kind' = 'sequence',
    'fields.id.start' = '0',
    'fields.id.end' = '3000'
);
create table sink (
    `id` int primary key,
    `data` string,
    `part` string
) partitioned by (`part`) with (
    'connector' = 'hudi',
    'path' = '/tmp/sink',
    'write.batch.size' = '0.001',  -- 1024 bytes
    'write.task.max.size' = '101.001',  -- 101.001MB
    'write.merge.max_memory' = '1'  -- 1024 bytes
);

insert into sink select `id`, `data`, concat('part', cast(`id` as string)) as `part` from source;