[Feature Request] Control the deletion frequency of S3ObjectControlManager$ObjectCleaner#clean

[superhx]

Who is this feature for?

Ops

What problem are they facing today?

When a Topic with a large amount of data shortens the TTL or is deleted, the concurrent deletion of a large number of Objects will cause S3 to be rate-limited.

Why is solving this impactful?

Proposed solution

Control deletion concurrency:

• For ordinary Objects, control 1000 per batch with a total of 8 concurrencies.
• For Composite Objects, DELETE / DEEP_DELETE ?

Additional notes

No response

[Arnab1996]

Hi @superhx, I am happy to take a swing at this if it hasn't been picked up.

/assign

I read through S3ObjectControlManager end-to-end and want to validate my understanding before writing code, since the right shape of this fix depends on a few architectural calls only you can make.

Looking at ObjectCleaner.batchDelete (around line 570), the shallow-delete path is actually somewhat bounded today, MAX_DELETE_BATCH_COUNT=2000 divided by the 1000-per-request batch size means each cleaning round caps at ~2 concurrent DeleteObjects calls. The bigger fan-out hazard sits in deepDelete (line 594): one CompositeObject.delete() per composite, each of which issues 2+ S3 calls of its own. With 2000 expired composites in a single round, in-flight S3 traffic is unbounded, that matches the rate-limit symptom you described.

Concrete proposal:

• Two new configs (in Config.java, plumbed through AutoMQConfig → KafkaConfig → ConfigUtils like objectRetentionTimeInSecond):
• s3.object.delete.max.concurrency (default 8, your suggestion)
• s3.composite.object.delete.max.concurrency (default 2 — composite is heavier, separating it lets ops tune the two paths independently)
• In ObjectCleaner, gate batchDelete with Semaphore.tryAcquire and reschedule on miss via the existing lifecycleCheckTimer. Avoiding acquire() is important since this runs on the controller event loop, I noticed the existing comment at line 470 about not blocking it.
• MAX_DELETE_BATCH_COUNT likely collapses into the new configs but I'd rather check first.

The questions I'd love your call on before I start:

1. Should MAX_DELETE_BATCH_COUNT (2000) merge into the new configs, or stay as a per-tick safety cap on top?
2. Where do you want the gate to live, at the S3ObjectControlManager layer (what I drafted), or do you want me to push it down into AwsObjectStorage's existing DeleteObjectsAccumulator, which already has a getMaxObjectStorageConcurrency()? Stacking two limits would be confusing.
3. For composites: separate config (what I drafted) or unified with a heavier "weight" multiplier per composite?

If the direction sounds right, I'll extend S3ObjectControlManagerTest with two new cases (one for shallow gating, one for composite gating, both using the same in-flight-counting mock pattern as testCheckS3ObjectsLifecycle_inflightLimit) and open a PR.

Thanks for the detailed issue, it was clear enough that I could trace the code path without having to ask basic questions, which I appreciate.