public static StorageInterface make(final PluginRegistry pluginRegistry,
                                        final String pluginId,
                                        final Map<String, Object> pluginConfiguration,
                                        final Validator validator) {
        Optional<Class<? extends StorageInterface>> optional = allStorageClasses(pluginRegistry)
            .filter(clazz -> Plugin.getId(clazz).map(id -> id.equalsIgnoreCase(pluginId)).orElse(false))
            .findFirst();

        if (optional.isEmpty()) {
            String storageIds = getLoggableStorageIds(pluginRegistry);
            throw new KestraRuntimeException(String.format(
                "No storage interface can be found for '%s=%s'. Supported types are: %s", KESTRA_STORAGE_TYPE_CONFIG, pluginId, storageIds
            ));
        }

        Class<? extends StorageInterface> pluginClass = optional.get();

        // Storage are handle as any serializable/deserialize plugins.
        StorageInterface plugin;
        try {
            // Make sure config is not null, otherwise deserialization result will be null too.
            Map<String, Object> nonEmptyConfig = Optional.ofNullable(pluginConfiguration).orElse(Map.of());
            plugin = JacksonMapper.toMap(nonEmptyConfig, pluginClass);
        } catch (Exception e) {
            throw new KestraRuntimeException(String.format(
                "Failed to create storage '%s'. Error: %s", pluginId, e.getMessage())
            );
        }

        // Validate configuration.
        Set<ConstraintViolation<StorageInterface>> violations;
        try {
            violations = validator.validate(plugin);
        } catch (ConstraintViolationException e) {
            throw new KestraRuntimeException(String.format(
                "Failed to validate configuration for storage '%s'. Error: %s", pluginId, e.getMessage())
            );
        }
        if (!violations.isEmpty()) {
            ConstraintViolationException e = new ConstraintViolationException(violations);
            throw new KestraRuntimeException(String.format(
                "Invalid configuration for storage '%s'. Error: '%s'", pluginId, e.getMessage()), e
            );
        }

        try {
            plugin.init();
        } catch (IOException e) {
            throw new KestraRuntimeException(String.format(
                "Failed to initialize storage '%s'. Error: %s", pluginId, e.getMessage()), e
            );
        }
        return plugin;
    }
@Test
    void shouldReturnStorageGivenValidId() {
        StorageInterface storage = StorageInterfaceFactory.make(registry, "local", Map.of("basePath", "/tmp/kestra"), validator);
        Assertions.assertNotNull(storage);
        Assertions.assertEquals(LocalStorage.class.getName(), storage.getType());
    }