Set workflowId for upload

Author: kowser-orkes

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

@@ -36,6 +36,7 @@
 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.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -381,7 +382,7 @@ private void executeTask(Worker worker, Task task) {
                     worker.getIdentity());
             result = worker.execute(task);
             stopwatch.stop();
-            uploadFilesToFileStorage(result);
+            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());
@@ -468,7 +469,7 @@ private void updateTaskResult(int count, Task task, TaskResult result, Worker wo
     }
 
     @VisibleForTesting
-    void uploadFilesToFileStorage(TaskResult result) {
+    void uploadFilesToFileStorage(TaskResult result, String workflowId, String taskId) {
         if (fileClient == null || result.getOutputData() == null) {
             return;
         }
@@ -478,7 +479,10 @@ void uploadFilesToFileStorage(TaskResult result) {
             }
             if (fh.getFileHandleId() == null) {
                 Path path = ((LocalFileHandler) fh).getPath();
-                FileHandler uploaded = fileClient.upload(path, fh.getContentType());
+                FileUploadOptions options = new FileUploadOptions()
+                        .setContentType(fh.getContentType())
+                        .setTaskId(taskId);
+                FileHandler uploaded = fileClient.upload(workflowId, path, options);
                 entry.setValue(uploaded.getFileHandleId());
             } else {
                 entry.setValue(fh.getFileHandleId());

conductor-client/src/main/java/org/conductoross/conductor/client/FileClient.java

@@ -30,6 +30,7 @@
 import com.netflix.conductor.client.http.ConductorClient;
 import com.netflix.conductor.client.http.ConductorClientRequest;
 import com.netflix.conductor.client.http.ConductorClientRequest.Method;
+import com.netflix.conductor.sdk.workflow.executor.task.TaskContext;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 
@@ -51,12 +52,18 @@
  */
 public class FileClient implements FileUploader {
 
-    private static final TypeReference<FileUploadResponse> FILE_UPLOAD_RESPONSE_TYPE = new TypeReference<>() {};
-    private static final TypeReference<FileUploadUrlResponse> FILE_UPLOAD_URL_RESPONSE_TYPE = new TypeReference<>() {};
-    private static final TypeReference<FileUploadCompleteResponse> FILE_UPLOAD_COMPLETE_RESPONSE_TYPE = new TypeReference<>() {};
-    private static final TypeReference<FileDownloadUrlResponse> FILE_DOWNLOAD_URL_RESPONSE_TYPE = new TypeReference<>() {};
-    private static final TypeReference<FileHandle> FILE_HANDLE_TYPE = new TypeReference<>() {};
-    private static final TypeReference<MultipartInitResponse> MULTIPART_INIT_RESPONSE_TYPE = new TypeReference<>() {};
+    private static final TypeReference<FileUploadResponse> FILE_UPLOAD_RESPONSE_TYPE = new TypeReference<>() {
+    };
+    private static final TypeReference<FileUploadUrlResponse> FILE_UPLOAD_URL_RESPONSE_TYPE = new TypeReference<>() {
+    };
+    private static final TypeReference<FileUploadCompleteResponse> FILE_UPLOAD_COMPLETE_RESPONSE_TYPE = new TypeReference<>() {
+    };
+    private static final TypeReference<FileDownloadUrlResponse> FILE_DOWNLOAD_URL_RESPONSE_TYPE = new TypeReference<>() {
+    };
+    private static final TypeReference<FileHandle> FILE_HANDLE_TYPE = new TypeReference<>() {
+    };
+    private static final TypeReference<MultipartInitResponse> MULTIPART_INIT_RESPONSE_TYPE = new TypeReference<>() {
+    };
 
     private final ConductorClient client;
     private final FileClientProperties properties;
@@ -73,8 +80,8 @@ public FileClient(ConductorClient client) {
     /**
      * Full-args constructor.
      *
-     * @param properties                         {@code null} to use defaults
-     * @param fileStorageBackendsByStorageType   {@code null} to use the built-in backends
+     * @param properties                       {@code null} to use defaults
+     * @param fileStorageBackendsByStorageType {@code null} to use the built-in backends
      */
     public FileClient(ConductorClient client, FileClientProperties properties, Map<StorageType, FileStorageBackend> fileStorageBackendsByStorageType) {
         this.client = client;
@@ -84,25 +91,24 @@ public FileClient(ConductorClient client, FileClientProperties properties, Map<S
 
     // --- FileUploader (developer-facing) ---
 
-    @Override
-    public FileHandler upload(Path localFile) {
-        return upload(localFile, "application/octet-stream");
-    }
-
     /**
-     * Uploads a local file. Creates the server-side metadata record, selects the backend that
-     * matches the server-reported {@link StorageType}, and uses multipart when the file size
-     * exceeds {@link FileClientProperties#getMultipartThreshold()} AND the backend supports
-     * multipart (see {@link FileStorageBackend#hasMultipartSupport()}).
+     * Uploads a local file with the given {@link FileUploadOptions}.
+     *
+     * <p>{@code workflowId} is required; pass {@code null} only if you are prepared to see a
+     * {@link FileStorageException}. When called from inside a worker and {@code options.taskId}
+     * is null, {@code taskId} is auto-filled from the active {@link TaskContext}.
      *
-     * @throws FileStorageException if no backend is registered for the server's storage type,
-     *         or if any step of the upload fails
+     * @throws FileStorageException if {@code workflowId} is null, if no backend is registered for
+     *                              the server's storage type, or if any step of the upload fails
      */
     @Override
-    public FileHandler upload(Path localFile, String contentType) {
+    public FileHandler upload(String workflowId, Path localFile, FileUploadOptions options) {
+        if (workflowId == null) {
+            throw new FileStorageException("workflowId is required");
+        }
         try {
-            long fileSize = Files.size(localFile);
-            FileUploadRequest request = FileHandlerConverter.toFileUploadRequest(localFile, contentType, fileSize);
+            fillDefaults(options, localFile);
+            FileUploadRequest request = FileHandlerConverter.toFileUploadRequest(workflowId, options);
 
             FileUploadResponse response = createFileOnServer(request);
 
@@ -112,7 +118,7 @@ public FileHandler upload(Path localFile, String contentType) {
                         + " but SDK only supports: " + fileStorageBackendsByStorageType.keySet());
             }
 
-            if (fileSize > properties.getMultipartThreshold() && storageBackend.hasMultipartSupport()) {
+            if (options.getFileSize() > properties.getMultipartThreshold() && storageBackend.hasMultipartSupport()) {
                 uploadMultipart(response.getFileHandleId(), response.getStorageType(), localFile);
             } else {
                 storageBackend.upload(response.getUploadUrl(), localFile);
@@ -125,22 +131,52 @@ public FileHandler upload(Path localFile, String contentType) {
         }
     }
 
+    /**
+     * Buffers the stream to a temp file, then delegates to {@link #upload(String, Path, FileUploadOptions)}.
+     */
     @Override
-    public FileHandler upload(InputStream inputStream) {
-        return upload(inputStream, "application/octet-stream");
-    }
-
-    @Override
-    public FileHandler upload(InputStream inputStream, String contentType) {
+    public FileHandler upload(String workflowId, InputStream inputStream, FileUploadOptions options) {
         try {
             Path temp = Files.createTempFile("conductor-upload-", ".tmp");
-            Files.copy(inputStream, temp, java.nio.file.StandardCopyOption.REPLACE_EXISTING);
-            return upload(temp, contentType);
+            Files.copy(inputStream, temp, StandardCopyOption.REPLACE_EXISTING);
+            return upload(workflowId, temp, options);
         } catch (IOException e) {
             throw new FileStorageException("Failed to write InputStream to temp file", e);
         }
     }
 
+    @Override
+    public FileHandler upload(String workflowId, Path localFile) {
+        return upload(workflowId, localFile, new FileUploadOptions());
+    }
+
+    @Override
+    public FileHandler upload(String workflowId, InputStream inputStream) {
+        return upload(workflowId, inputStream, new FileUploadOptions());
+    }
+
+    /**
+     * Populates unset options from the local file and active {@link TaskContext}. Mutates
+     * {@code options} in place — SDK-internal helper only.
+     */
+    private static void fillDefaults(FileUploadOptions options, Path localFile) throws IOException {
+        if (options.getFileName() == null) {
+            options.setFileName(localFile.getFileName().toString());
+        }
+        if (options.getFileSize() == 0) {
+            options.setFileSize(Files.size(localFile));
+        }
+        if (options.getContentType() == null) {
+            options.setContentType("application/octet-stream");
+        }
+        if (options.getTaskId() == null) {
+            TaskContext ctx = TaskContext.get();
+            if (ctx != null) {
+                options.setTaskId(ctx.getTaskId());
+            }
+        }
+    }
+
     // --- SDK-internal (public for cross-package access, not developer-facing) ---
 
     /**
@@ -158,7 +194,9 @@ public void download(String workflowId, String fileHandleId, StorageType storage
         fileStorageBackendsByStorageType.get(storageType).download(urlResponse.getDownloadUrl(), destination);
     }
 
-    /** Fetches the {@link FileHandle} metadata via {@code GET /api/files/{fileId}}. */
+    /**
+     * Fetches the {@link FileHandle} metadata via {@code GET /api/files/{fileId}}.
+     */
     public FileHandle getMetadata(String fileHandleId) {
         ConductorClientRequest request = ConductorClientRequest.builder()
                 .method(Method.GET)

conductor-client/src/main/java/org/conductoross/conductor/sdk/file/FileHandlerConverter.java

@@ -15,7 +15,6 @@
 import java.nio.file.Path;
 
 import org.conductoross.conductor.client.FileClient;
-import org.conductoross.conductor.client.model.file.FileHandle;
 import org.conductoross.conductor.client.model.file.FileUploadRequest;
 import org.conductoross.conductor.client.model.file.FileUploadResponse;
 
@@ -38,21 +37,14 @@ public static ManagedFileHandler toManagedFileHandler(FileUploadResponse respons
         return handler;
     }
 
-    /** Copies server-provided metadata onto an existing {@link ManagedFileHandler}. */
-    public static void populateFromMetadata(ManagedFileHandler handler, FileHandle metadata) {
-        handler.setFileName(metadata.getFileName());
-        handler.setContentType(metadata.getContentType());
-        handler.setFileSize(metadata.getFileSize());
-        handler.setStorageType(metadata.getStorageType());
-    }
-
     /** Builds the {@link FileUploadRequest} payload for {@code POST /api/files}. */
-    public static FileUploadRequest toFileUploadRequest(Path localFile, String contentType,
-            long fileSize) {
+    public static FileUploadRequest toFileUploadRequest(String workflowId, FileUploadOptions options) {
         FileUploadRequest request = new FileUploadRequest();
-        request.setFileName(localFile.getFileName().toString());
-        request.setContentType(contentType);
-        request.setFileSize(fileSize);
+        request.setFileName(options.getFileName());
+        request.setContentType(options.getContentType());
+        request.setFileSize(options.getFileSize());
+        request.setWorkflowId(workflowId);
+        request.setTaskId(options.getTaskId());
         return request;
     }
 }

conductor-client/src/main/java/org/conductoross/conductor/sdk/file/FileUploadOptions.java

@@ -0,0 +1,72 @@
+/*
+ * 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 org.conductoross.conductor.sdk.file;
+
+/**
+ * Optional metadata to attach to a file upload. {@code workflowId} is supplied as the explicit
+ * first argument on {@link FileUploader#upload}; this type carries everything else.
+ *
+ * <p>When uploading inside a worker, {@code taskId} is populated automatically from the active
+ * {@link com.netflix.conductor.sdk.workflow.executor.task.TaskContext} if not set here.
+ * Providing it explicitly overrides the auto-detected value.
+ *
+ * <pre>{@code
+ * FileUploadOptions options = new FileUploadOptions()
+ *         .setContentType("application/pdf")
+ *         .setTaskId("task-456");
+ * FileHandler handler = fileUploader.upload(workflowId, path, options);
+ * }</pre>
+ */
+public class FileUploadOptions {
+
+    private String taskId;
+    private String fileName;
+    private String contentType;
+    private long fileSize;
+
+    public String getTaskId() {
+        return taskId;
+    }
+
+    public FileUploadOptions setTaskId(String taskId) {
+        this.taskId = taskId;
+        return this;
+    }
+
+    public String getFileName() {
+        return fileName;
+    }
+
+    public FileUploadOptions setFileName(String fileName) {
+        this.fileName = fileName;
+        return this;
+    }
+
+    public String getContentType() {
+        return contentType;
+    }
+
+    public FileUploadOptions setContentType(String contentType) {
+        this.contentType = contentType;
+        return this;
+    }
+
+    public long getFileSize() {
+        return fileSize;
+    }
+
+    public FileUploadOptions setFileSize(long fileSize) {
+        this.fileSize = fileSize;
+        return this;
+    }
+}

conductor-client/src/main/java/org/conductoross/conductor/sdk/file/FileUploader.java

@@ -24,18 +24,20 @@
  * populated with the prefixed handle {@code conductor://file/<fileId>}. The {@link InputStream}
  * overloads stream the content to a temporary file first, then delegate to the {@link Path}
  * overload.
+ *
+ * <p>When called from within a worker, {@code workflowId} and {@code taskId} are injected
+ * automatically from the active task context. Use {@link FileUploadOptions} to override them or
+ * to supply additional metadata when uploading outside a worker.
  */
 public interface FileUploader {
 
-    /** Upload from an {@link InputStream} with content type {@code application/octet-stream}. */
-    FileHandler upload(InputStream inputStream);
+    FileHandler upload(String workflowId, Path localFile);
 
-    /** Upload from an {@link InputStream} with an explicit content type. */
-    FileHandler upload(InputStream inputStream, String contentType);
+    FileHandler upload(String workflowId, InputStream inputStream);
 
-    /** Upload from a local file with content type {@code application/octet-stream}. */
-    FileHandler upload(Path localFile);
+    /** Upload from a local file with the given options. */
+    FileHandler upload(String workflowId, Path localFile, FileUploadOptions options);
 
-    /** Upload from a local file with an explicit content type. */
-    FileHandler upload(Path localFile, String contentType);
+    /** Upload from an {@link InputStream} with the given options. */
+    FileHandler upload(String workflowId, InputStream inputStream, FileUploadOptions options);
 }

conductor-client/src/main/java/org/conductoross/conductor/sdk/file/LocalFileHandler.java

@@ -23,7 +23,7 @@
  * {@link FileHandler} wrapping a local file that has not been uploaded.
  * {@link #getFileHandleId()} returns {@code null}. Returned by
  * {@link FileHandler#fromLocalFile(Path)}; consumed by
- * {@link org.conductoross.conductor.client.FileClient#upload(Path)} or by
+ * {@link FileClient#upload(String, Path)} or by
  * {@code TaskRunner}'s auto-upload interception on task completion.
  */
 public class LocalFileHandler implements FileHandler {