Repository: kestra-io/kestra
File: core/src/main/java/io/kestra/core/utils/NamespaceFilesUtils.java (Java)

Code before:
        stopWatch.start();

        List<String> include = runContext.render(namespaceFiles.getInclude()).asList(String.class);
        List<String> exclude = runContext.render(namespaceFiles.getExclude()).asList(String.class);
        FileExistComportment fileExistComportment = runContext.render(namespaceFiles.getIfExists())
            .as(FileExistComportment.class).orElse(FileExistComportment.OVERWRITE);
        List<String> namespaces = runContext.render(namespaceFiles.getNamespaces()).asList(String.class);
        Boolean folderPerNamespace = runContext.render(namespaceFiles.getFolderPerNamespace()).as(Boolean.class)
            .orElse(false);

        List<NamespaceFile> matchedNamespaceFiles = new ArrayList<>();
        for (String namespace : namespaces) {
            List<NamespaceFile> files = runContext.storage()
                .namespace(namespace)
                .findAllFilesMatching(include, exclude);

          matchedNamespaceFiles.addAll(files);
        }

        int parallelism = Math.max(Runtime.getRuntime().availableProcessors() * 4, 32);
        Flux.fromIterable(matchedNamespaceFiles)
            .parallel(parallelism)
            .runOn(Schedulers.fromExecutorService(executorService))
            .doOnNext(throwConsumer(nsFile -> {
                InputStream content = runContext.storage().getFile(nsFile.uri());
                Path path = folderPerNamespace ?
                    Path.of(nsFile.namespace() + "/" + nsFile.path()) :
                    Path.of(nsFile.path());
                runContext.workingDir().putFile(path, content, fileExistComportment);
            }))
            .sequential()
            .blockLast();

        Duration duration = stopWatch.getDuration();

        runContext.metric(Counter.of("namespacefiles.count", matchedNamespaceFiles.size()));
        runContext.metric(Timer.of("namespacefiles.duration", duration));

        runContext.logger().info("Loaded {} namespace files from '{}' in {}",
            matchedNamespaceFiles.size(),
            StringUtils.join(namespaces, ", "),
            DurationFormatUtils.formatDurationHMS(duration.toMillis())
        );
    }
}


Review comment (performance):
I'm a bit worry to allow using all threads by a single task, this may slow down the dowload of concurrently processing tasks as I don't know how fair is the reactor scheduler.

We may only allocate half threads to a single task upload to avoid that.

Code after:
        throws Exception {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();

        List<String> include = runContext.render(namespaceFiles.getInclude()).asList(String.class);
        List<String> exclude = runContext.render(namespaceFiles.getExclude()).asList(String.class);
        FileExistComportment fileExistComportment = runContext.render(namespaceFiles.getIfExists())
            .as(FileExistComportment.class).orElse(FileExistComportment.OVERWRITE);
        List<String> namespaces = runContext.render(namespaceFiles.getNamespaces()).asList(String.class);
        Boolean folderPerNamespace = runContext.render(namespaceFiles.getFolderPerNamespace()).as(Boolean.class)
            .orElse(false);

        List<NamespaceFile> matchedNamespaceFiles = new ArrayList<>();
        for (String namespace : namespaces) {
            List<NamespaceFile> files = runContext.storage()
                .namespace(namespace)
                .findAllFilesMatching(include, exclude);

          matchedNamespaceFiles.addAll(files);
        }

        // Use half of the available threads to avoid impacting concurrent tasks
        int parallelism = maxThreads / 2;
        Flux.fromIterable(matchedNamespaceFiles)
            .parallel(parallelism)
            .runOn(Schedulers.fromExecutorService(executorService))
            .doOnNext(throwConsumer(nsFile -> {
                InputStream content = runContext.storage().getFile(nsFile.uri());
                Path path = folderPerNamespace ?
                    Path.of(nsFile.namespace() + "/" + nsFile.path()) :
                    Path.of(nsFile.path());
                runContext.workingDir().putFile(path, content, fileExistComportment);
            }))
            .sequential()
            .blockLast();

        Duration duration = stopWatch.getDuration();

        runContext.metric(Counter.of("namespacefiles.count", matchedNamespaceFiles.size()));
        runContext.metric(Timer.of("namespacefiles.duration", duration));

        runContext.logger().info("Loaded {} namespace files from '{}' in {}",
            matchedNamespaceFiles.size(),
            StringUtils.join(namespaces, ", "),
            DurationFormatUtils.formatDurationHMS(duration.toMillis())
        );
    }
}