ExternalPayloadStorageUtils { public <T> void verifyAndUpload(T entity, PayloadType payloadType) { long threshold = 0L; long maxThreshold = 0L; Map<String, Object> payload = new HashMap<>(); String workflowId = ""; switch (payloadType) { case TASK_INPUT: threshold = configuration.getTaskInputPayloadSizeThresholdKB(); maxThreshold = configuration.getMaxTaskInputPayloadSizeThresholdKB(); payload = ((Task) entity).getInputData(); workflowId = ((Task) entity).getWorkflowInstanceId(); break; case TASK_OUTPUT: threshold = configuration.getTaskOutputPayloadSizeThresholdKB(); maxThreshold = configuration.getMaxTaskOutputPayloadSizeThresholdKB(); payload = ((Task) entity).getOutputData(); workflowId = ((Task) entity).getWorkflowInstanceId(); break; case WORKFLOW_INPUT: threshold = configuration.getWorkflowInputPayloadSizeThresholdKB(); maxThreshold = configuration.getMaxWorkflowInputPayloadSizeThresholdKB(); payload = ((Workflow) entity).getInput(); workflowId = ((Workflow) entity).getWorkflowId(); break; case WORKFLOW_OUTPUT: threshold = configuration.getWorkflowOutputPayloadSizeThresholdKB(); maxThreshold = configuration.getMaxWorkflowOutputPayloadSizeThresholdKB(); payload = ((Workflow) entity).getOutput(); workflowId = ((Workflow) entity).getWorkflowId(); break; } try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) { objectMapper.writeValue(byteArrayOutputStream, payload); byte[] payloadBytes = byteArrayOutputStream.toByteArray(); long payloadSize = payloadBytes.length; if (payloadSize > maxThreshold * 1024) { if (entity instanceof Task) { String errorMsg = String.format("The payload size: %dB of task: %s in workflow: %s is greater than the permissible limit: %dKB", payloadSize, ((Task) entity).getTaskId(), ((Task) entity).getWorkflowInstanceId(), maxThreshold); failTask(((Task) entity), payloadType, errorMsg); } else { String errorMsg = String.format("The output payload size: %dB of workflow: %s is greater than the permissible limit: %dKB", payloadSize, ((Workflow) entity).getWorkflowId(), maxThreshold); failWorkflow(errorMsg); } } else if (payloadSize > threshold * 1024) { switch (payloadType) { case TASK_INPUT: ((Task) entity).setInputData(null); ((Task) entity).setExternalInputPayloadStoragePath(uploadHelper(payloadBytes, payloadSize, PayloadType.TASK_INPUT)); Monitors.recordExternalPayloadStorageUsage(((Task) entity).getTaskDefName(), ExternalPayloadStorage.Operation.WRITE.toString(), PayloadType.TASK_INPUT.toString()); break; case TASK_OUTPUT: ((Task) entity).setOutputData(null); ((Task) entity).setExternalOutputPayloadStoragePath(uploadHelper(payloadBytes, payloadSize, PayloadType.TASK_OUTPUT)); Monitors.recordExternalPayloadStorageUsage(((Task) entity).getTaskDefName(), ExternalPayloadStorage.Operation.WRITE.toString(), PayloadType.TASK_OUTPUT.toString()); break; case WORKFLOW_INPUT: ((Workflow) entity).setInput(null); ((Workflow) entity).setExternalInputPayloadStoragePath(uploadHelper(payloadBytes, payloadSize, PayloadType.WORKFLOW_INPUT)); Monitors.recordExternalPayloadStorageUsage(((Workflow) entity).getWorkflowName(), ExternalPayloadStorage.Operation.WRITE.toString(), PayloadType.WORKFLOW_INPUT.toString()); break; case WORKFLOW_OUTPUT: ((Workflow) entity).setOutput(null); ((Workflow) entity).setExternalOutputPayloadStoragePath(uploadHelper(payloadBytes, payloadSize, PayloadType.WORKFLOW_OUTPUT)); Monitors.recordExternalPayloadStorageUsage(((Workflow) entity).getWorkflowName(), ExternalPayloadStorage.Operation.WRITE.toString(), PayloadType.WORKFLOW_OUTPUT.toString()); break; } } } catch (IOException e) { logger.error("Unable to upload payload to external storage for workflow: {}", workflowId, e); throw new ApplicationException(ApplicationException.Code.INTERNAL_ERROR, e); } } @Inject  ExternalPayloadStorageUtils(ExternalPayloadStorage externalPayloadStorage,
                                       Configuration configuration,
                                       ObjectMapper objectMapper); @SuppressWarnings("unchecked") Map<String, Object> downloadPayload(String path); void verifyAndUpload(T entity, PayloadType payloadType);  }
@SuppressWarnings("unchecked") @Test public void testUploadTaskPayload() throws IOException { AtomicInteger uploadCount = new AtomicInteger(0); InputStream stream = ExternalPayloadStorageUtilsTest.class.getResourceAsStream("/payload.json"); Map<String, Object> payload = objectMapper.readValue(stream, Map.class); when(externalPayloadStorage.getLocation(ExternalPayloadStorage.Operation.WRITE, ExternalPayloadStorage.PayloadType.TASK_INPUT, "")).thenReturn(location); doAnswer(invocation -> { uploadCount.incrementAndGet(); return null; }).when(externalPayloadStorage).upload(anyString(), any(), anyLong()); Task task = new Task(); task.setInputData(payload); externalPayloadStorageUtils.verifyAndUpload(task, ExternalPayloadStorage.PayloadType.TASK_INPUT); assertNull(task.getInputData()); assertEquals(1, uploadCount.get()); assertNotNull(task.getExternalInputPayloadStoragePath()); }
@SuppressWarnings("unchecked") @Test public void testUploadWorkflowPayload() throws IOException { AtomicInteger uploadCount = new AtomicInteger(0); InputStream stream = ExternalPayloadStorageUtilsTest.class.getResourceAsStream("/payload.json"); Map<String, Object> payload = objectMapper.readValue(stream, Map.class); when(externalPayloadStorage.getLocation(ExternalPayloadStorage.Operation.WRITE, ExternalPayloadStorage.PayloadType.WORKFLOW_OUTPUT, "")).thenReturn(location); doAnswer(invocation -> { uploadCount.incrementAndGet(); return null; }).when(externalPayloadStorage).upload(anyString(), any(), anyLong()); Workflow workflow = new Workflow(); workflow.setOutput(payload); externalPayloadStorageUtils.verifyAndUpload(workflow, ExternalPayloadStorage.PayloadType.WORKFLOW_OUTPUT); assertNull(workflow.getOutput()); assertEquals(1, uploadCount.get()); assertNotNull(workflow.getExternalOutputPayloadStoragePath()); }