Hey 👋 , Running flink 1.16 with rocksdb incremental checkpoint We have an app that has ~5TB of state and to help scale it, we decided to double the task parallelism which led to increase state size to increase by 50%. I think that is primarily because existing subtasks didn’t delete now moved subtasks since our state TTL is configured as

StateTtlConfig
      .newBuilder(Time.seconds(retentionSeconds))
      .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
      .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
      .cleanupInRocksdbCompactFilter(10000)
      .build();

Which I assume means they will be deleted during the next rocksdb compaction. Flink by default triggers compaction every 30 days and I think until flink 1.18 there isn’t user configuration to change that. Our bigger concern is that rocksdb compaction for the lowermost level will not be triggered, we are still away from the size limit of L4. Does anyone have recommendation on changing our configuration to reduce the state size for faster recovery and smaller checkpoints.

Level    Files   Size     Score 
--------------------------------
  L0      2/0    5.53 MB   0.5      
  L1      4/0   202.85 MB   0.8    
  L2     41/0    2.46 GB   1.0      
  L3    206/0    9.60 GB   0.4     
  L4    964/0   59.38 GB   0.2