Repository: kestra-io/kestra
File: core/src/main/java/io/kestra/core/storages/kv/KVPurgeCleaner.java (Java)

Code before:
import io.micronaut.scheduling.annotation.Scheduled;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
import java.io.IOException;
import java.time.Instant;
import java.util.List;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Requires(property = "kestra.kv.purge-expired.enabled", value = "true", defaultValue = "true")
@Singleton
public class KVPurgeCleaner {

    @Inject
    private KVStoreService kvStoreService;

    @Inject
    private FlowRepositoryInterface flowRepository;

    @Inject
    private KvMetadataRepositoryInterface kvMetadataRepository;

    @Value("${kestra.kv.purge-expired.batch-size:1000}")
    private Integer batchSize;

    @Scheduled(initialDelay = "${kestra.kv.purge-expired.initial-delay:PT6H}", fixedDelay = "${kestra.kv.purge-expired.fixed-delay:PT6H}")
    public  void purgeExpired(){
        log.info("Start cleaning expired KV store entries");
        List<String> tenants = findTenants();
        for (String tenant : tenants) {
            purgeKVEntriesForTenant(tenant);
        }
    }

    private void purgeKVEntriesForTenant(String tenant) {
        List<String> namespaces = findNamespaces(tenant);
        Instant now = Instant.now();
        for (String namespace : namespaces) {
            try {
                KVStore kvStore = kvStoreService.get(tenant, namespace, namespace);
                List<KVEntry> expiredEntries;
                do {
                    expiredEntries = kvMetadataRepository.find(
                            //We always fetch the first page because we delete every KV entries that we find.
                            Pageable.from(1, batchSize, Sort.of(Order.asc("name"))),
                            tenant,
                            List.of(
                                QueryFilter.builder().field(Field.NAMESPACE).value(namespace).operation(Op.EQUALS).build(),
                                QueryFilter.builder().field(Field.EXPIRATION_DATE).value(now).operation(Op.LESS_THAN).build()
                            ),
                            false,

Review comment (suggestion):
```suggestion
    @Scheduled(initialDelay = "${kestra.kv.purge-expired.initial-delay:PT1H}", fixedDelay = "${kestra.kv.purge-expired.fixed-delay:PT1H}")
```

Nitpick but most of our internal purge mechanisms are using 1H delay so better to align here

Code after:
import io.micronaut.scheduling.annotation.Scheduled;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
import java.io.IOException;
import java.time.Instant;
import java.util.List;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Requires(property = "kestra.kv.purge-expired.enabled", value = "true", defaultValue = "true")
@Singleton
public class KVPurgeCleaner {

    @Inject
    private KVStoreService kvStoreService;

    @Inject
    private FlowRepositoryInterface flowRepository;

    @Inject
    private KvMetadataRepositoryInterface kvMetadataRepository;

    @Value("${kestra.kv.purge-expired.batch-size:1000}")
    private Integer batchSize;

    @Scheduled(initialDelay = "${kestra.kv.purge-expired.initial-delay:PT1H}", fixedDelay = "${kestra.kv.purge-expired.fixed-delay:PT1H}")
    public  void purgeExpired(){
        log.info("Start cleaning expired KV store entries");
        List<String> tenants = findTenants();
        for (String tenant : tenants) {
            purgeKVEntriesForTenant(tenant);
        }
    }

    private void purgeKVEntriesForTenant(String tenant) {
        List<String> namespaces = findNamespaces(tenant);
        Instant now = Instant.now();
        for (String namespace : namespaces) {
            try {
                KVStore kvStore = kvStoreService.get(tenant, namespace, namespace);
                List<KVEntry> expiredEntries;
                do {
                    expiredEntries = kvMetadataRepository.find(
                            //We always fetch the first page because we delete every KV entries that we find.
                            Pageable.from(1, batchSize, Sort.of(Order.asc("name"))),
                            tenant,
                            List.of(
                                QueryFilter.builder().field(Field.NAMESPACE).value(namespace).operation(Op.EQUALS).build(),
                                QueryFilter.builder().field(Field.EXPIRATION_DATE).value(now).operation(Op.LESS_THAN).build()
                            ),
                            false,