File Storage Support for Java SDK #99

File Storage Support for Java SDK

Author: kowser-orkes

conductor-client-spring/build.gradle

@@ -20,6 +20,8 @@ repositories {
 dependencies {
     api project(":conductor-client")
     implementation 'org.springframework.boot:spring-boot-starter:3.5.13'
+
+    testImplementation 'org.springframework.boot:spring-boot-starter-test:3.3.13'
 }
 
 java {

conductor-client-spring/src/main/java/com/netflix/conductor/client/spring/ConductorClientAutoConfiguration.java

@@ -18,10 +18,13 @@
 import java.util.Optional;
 
 import org.apache.commons.lang3.StringUtils;
+import org.conductoross.conductor.client.FileClient;
+import org.conductoross.conductor.client.FileClientProperties;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.AutoConfigureOrder;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.core.Ordered;
@@ -82,6 +85,7 @@ public TaskRunnerConfigurer taskRunnerConfigurer(Environment env,
                                                      TaskClient taskClient,
                                                      ClientProperties clientProperties,
                                                      List<Worker> workers,
+                                                     Optional<FileClient> fileClient,
                                                      Optional<MetricsCollector> metricsCollector) {
         Map<String, Integer> taskThreadCount = new HashMap<>();
         for (Worker worker : workers) {
@@ -105,6 +109,7 @@ public TaskRunnerConfigurer taskRunnerConfigurer(Environment env,
                 .withTaskToDomain(clientProperties.getTaskToDomain())
                 .withShutdownGracePeriodSeconds(clientProperties.getShutdownGracePeriodSeconds())
                 .withTaskPollTimeout(clientProperties.getTaskPollTimeout());
+        fileClient.ifPresent(builder::withFileClient);
         metricsCollector.ifPresent(builder::withMetricsCollector);
         return builder.build();
     }
@@ -122,4 +127,19 @@ public WorkflowExecutor workflowExecutor(ConductorClient client, AnnotatedWorker
     public WorkflowClient workflowClient(ConductorClient client) {
         return new WorkflowClient(client);
     }
+
+    @Bean
+    @ConfigurationProperties("conductor.file-client")
+    @ConditionalOnBean(ConductorClient.class)
+    @ConditionalOnMissingBean
+    public FileClientProperties fileClientProperties() {
+        return new FileClientProperties();
+    }
+
+    @Bean
+    @ConditionalOnBean(ConductorClient.class)
+    @ConditionalOnMissingBean
+    public FileClient fileClient(ConductorClient client, FileClientProperties fileClientProperties) {
+        return new FileClient(client, fileClientProperties, null);
+    }
 }

conductor-client-spring/src/main/java/io/orkes/conductor/client/spring/OrkesConductorClientAutoConfiguration.java

@@ -13,9 +13,12 @@
 package io.orkes.conductor.client.spring;
 
 import org.apache.commons.lang3.StringUtils;
+import org.conductoross.conductor.client.FileClient;
+import org.conductoross.conductor.client.FileClientProperties;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
@@ -128,4 +131,18 @@ public SecretClient orkesSecretClient(OrkesClients clients) {
         return clients.getSecretClient();
     }
 
+    @Bean
+    @ConfigurationProperties("conductor.file-client")
+    @ConditionalOnBean(ApiClient.class)
+    @ConditionalOnMissingBean
+    public FileClientProperties fileClientProperties() {
+        return new FileClientProperties();
+    }
+
+    @Bean
+    @ConditionalOnBean(ApiClient.class)
+    @ConditionalOnMissingBean
+    public FileClient fileClient(ApiClient client, FileClientProperties fileClientProperties) {
+        return new FileClient(client, fileClientProperties, null);
+    }
 }

conductor-client-spring/src/test/java/com/netflix/conductor/client/spring/ConductorClientAutoConfigurationTest.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2026 Conductor Authors.
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.conductor.client.spring;
+
+import org.conductoross.conductor.client.FileClient;
+import org.conductoross.conductor.client.FileClientProperties;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ConductorClientAutoConfigurationTest {
+
+    private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(ConductorClientAutoConfiguration.class));
+
+    @Test
+    void zeroConfigYieldsWorkingFileClientBean() {
+        contextRunner
+                .withPropertyValues("conductor.client.base-path=http://localhost:8080/api")
+                .run(context -> {
+                    assertThat(context).hasSingleBean(FileClient.class);
+                    FileClientProperties properties = context.getBean(FileClientProperties.class);
+                    assertThat(properties.getLocalCacheDirectory())
+                            .startsWith(System.getProperty("java.io.tmpdir"));
+                });
+    }
+
+    @Test
+    void cacheDirectoryOverrideIsRespected() {
+        contextRunner
+                .withPropertyValues(
+                        "conductor.client.base-path=http://localhost:8080/api",
+                        "conductor.file-client.local-cache-directory=/tmp/custom-cache")
+                .run(context -> {
+                    FileClientProperties properties = context.getBean(FileClientProperties.class);
+                    assertThat(properties.getLocalCacheDirectory()).isEqualTo("/tmp/custom-cache");
+                });
+    }
+
+}

conductor-client/src/main/java/com/netflix/conductor/client/automator/TaskRunner.java

@@ -14,6 +14,7 @@
 
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedList;
@@ -34,6 +35,11 @@
 import java.util.function.Function;
 
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.conductoross.conductor.client.FileClient;
+import org.conductoross.conductor.sdk.file.FileHandler;
+import org.conductoross.conductor.sdk.file.FileUploadOptions;
+import org.conductoross.conductor.sdk.file.LocalFileHandler;
+import org.conductoross.conductor.sdk.file.WorkflowFileClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,6 +64,7 @@
 import com.netflix.conductor.common.metadata.tasks.Task;
 import com.netflix.conductor.common.metadata.tasks.TaskResult;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Stopwatch;
 import com.google.common.util.concurrent.Uninterruptibles;
 
@@ -80,6 +87,7 @@ class TaskRunner {
     private final EventDispatcher<TaskRunnerEvent> eventDispatcher;
     private final LinkedBlockingQueue<Task> tasksTobeExecuted;
     private final boolean enableUpdateV2;
+    private final FileClient fileClient;
     private static final int LEASE_EXTEND_RETRY_COUNT = 3;
     private static final double LEASE_EXTEND_DURATION_FACTOR = 0.8;
     private final ScheduledExecutorService leaseExtendExecutorService;
@@ -95,9 +103,11 @@ class TaskRunner {
                int taskPollTimeout,
                List<PollFilter> pollFilters,
                EventDispatcher<TaskRunnerEvent> eventDispatcher,
-               boolean useVirtualThreads) {
+               boolean useVirtualThreads,
+               FileClient fileClient) {
         this.worker = worker;
         this.taskClient = taskClient;
+        this.fileClient = fileClient;
         this.updateRetryCount = updateRetryCount;
         this.taskPollTimeout = taskPollTimeout;
         this.pollingIntervalInMillis = worker.getPollingInterval();
@@ -376,6 +386,11 @@ private void executeTask(Worker worker, Task task) {
             return;
         }
 
+        // Set FileClient on task for file storage support
+        if (fileClient != null) {
+            task.setWorkflowFileClient(new WorkflowFileClient(fileClient, task.getWorkflowInstanceId()));
+        }
+
         // Calculate  inbound network latency
         try {
             if(task.getExecutionMetadata().getServerSendTime() != null ){
@@ -398,6 +413,7 @@ private void executeTask(Worker worker, Task task) {
                     worker.getIdentity());
             result = worker.execute(task);
             stopwatch.stop();
+            uploadFilesToFileStorage(result, task.getWorkflowInstanceId(), task.getTaskId());
             eventDispatcher.publish(new TaskExecutionCompleted(taskType, task.getTaskId(), worker.getIdentity(), stopwatch.elapsed(TimeUnit.MILLISECONDS)));
             // record execution end time in task
             task.getExecutionMetadata().setExecutionEndTime(System.currentTimeMillis());
@@ -499,6 +515,26 @@ private void updateTaskResult(int count, Task task, TaskResult result, Worker wo
         }
     }
 
+    @VisibleForTesting
+    void uploadFilesToFileStorage(TaskResult result, String workflowId, String taskId) {
+        if (fileClient == null || result.getOutputData() == null) {
+            return;
+        }
+        for (var entry : result.getOutputData().entrySet()) {
+            if (!(entry.getValue() instanceof FileHandler fh)) {
+                continue;
+            }
+            if (fh.getFileHandleId() == null) {
+                Path path = ((LocalFileHandler) fh).getPath();
+                FileUploadOptions options = new FileUploadOptions()
+                        .setContentType(fh.getContentType())
+                        .setTaskId(taskId);
+                FileHandler uploaded = fileClient.upload(workflowId, path, options);
+                entry.setValue(uploaded);
+            }
+        }
+    }
+
     private Optional<String> upload(TaskResult result, String taskType) {
         try {
             return taskClient.evaluateAndUploadLargePayload(result.getOutputData(), taskType);

conductor-client/src/main/java/com/netflix/conductor/client/automator/TaskRunnerConfigurer.java

@@ -21,6 +21,7 @@
 import java.util.function.Consumer;
 
 import org.apache.commons.lang3.concurrent.BasicThreadFactory;
+import org.conductoross.conductor.client.FileClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,6 +56,7 @@ public class TaskRunnerConfigurer {
     private final EventDispatcher<TaskRunnerEvent> eventDispatcher;
     private final MetricsCollector metricsCollector;
     private final boolean useVirtualThreads;
+    private final FileClient fileClient;
 
     /**
      * @see TaskRunnerConfigurer.Builder
@@ -77,6 +79,7 @@ private TaskRunnerConfigurer(TaskRunnerConfigurer.Builder builder) {
         this.eventDispatcher = builder.eventDispatcher;
         this.metricsCollector = builder.metricsCollector;
         this.useVirtualThreads = builder.useVirtualThreads;
+        this.fileClient = builder.fileClient;
         builder.workers.forEach(this.workers::add);
         taskRunners = new LinkedList<>();
     }
@@ -179,7 +182,8 @@ private void startWorker(Worker worker) {
                 taskPollTimeout,
                 pollFilters,
                 eventDispatcher,
-                useVirtualThreads);
+                useVirtualThreads,
+                fileClient);
         // startWorker(worker) is executed by several threads.
         // taskRunners.add(taskRunner) without synchronization could lead to a race condition and unpredictable behavior,
         // including potential null values being inserted or corrupted state.
@@ -212,6 +216,7 @@ public static class Builder {
         private final List<PollFilter> pollFilters = new LinkedList<>();
         private final EventDispatcher<TaskRunnerEvent> eventDispatcher = new EventDispatcher<>();
         private boolean useVirtualThreads;
+        private FileClient fileClient;
 
         /**
          * Returns the event dispatcher used by this builder, allowing direct
@@ -382,5 +387,10 @@ public Builder withUseVirtualThreads(boolean useVirtualThreads) {
             this.useVirtualThreads = useVirtualThreads;
             return this;
         }
+
+        public Builder withFileClient(FileClient fileClient) {
+            this.fileClient = fileClient;
+            return this;
+        }
     }
 }

conductor-client/src/main/java/com/netflix/conductor/common/metadata/tasks/Task.java

@@ -18,10 +18,16 @@
 import java.util.Optional;
 
 import org.apache.commons.lang3.StringUtils;
+import org.conductoross.conductor.sdk.file.FileHandler;
+import org.conductoross.conductor.sdk.file.FileStorageException;
+import org.conductoross.conductor.sdk.file.FileUploader;
+import org.conductoross.conductor.sdk.file.ManagedFileHandler;
+import org.conductoross.conductor.sdk.file.WorkflowFileClient;
 
 import com.netflix.conductor.common.metadata.workflow.WorkflowTask;
 import com.netflix.conductor.common.run.tasks.TypedTask;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import lombok.*;
 
 @Data
@@ -179,6 +185,32 @@ public boolean isRetriable() {
 
     private long firstStartTime;
 
+    @JsonIgnore
+    private transient WorkflowFileClient workflowFileClient;
+
+    @JsonIgnore
+    public FileUploader getFileUploader() {
+        return workflowFileClient;
+    }
+
+    public void setWorkflowFileClient(WorkflowFileClient workflowFileClient) {
+        this.workflowFileClient = workflowFileClient;
+    }
+
+    public FileHandler getInputFileHandler(String key) {
+        if (workflowFileClient == null) {
+            throw new FileStorageException("FileClient is not configured; cannot access file input for key '" + key + "'");
+        }
+        Object value = getInputData().get(key);
+        String fileHandleId = FileHandler.extractFileHandleId(value);
+        if (FileHandler.isFileHandleId(fileHandleId)) {
+            return new ManagedFileHandler(fileHandleId, workflowFileClient);
+        }
+        throw new FileStorageException(
+                "Expected " + FileHandler.PREFIX
+                        + " reference for key '" + key + "', got: " + value);
+    }
+
     public void setInputData(Map<String, Object> inputData) {
         if (inputData == null) {
             inputData = new HashMap<>();