Iterate collection for unregistered Writeable in streams

build.gradle

@@ -151,10 +151,12 @@ dependencies {
     implementation 'org.junit.jupiter:junit-jupiter:5.10.1'
     implementation "com.google.guava:guava:33.0.0-jre"
     api group: 'org.opensearch', name:'opensearch-ml-client', version: "${opensearch_build}"
+    api group: 'org.opensearch.client', name: 'opensearch-rest-client', version: "${opensearch_version}"
     implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.14.0'
     implementation "org.opensearch:common-utils:${common_utils_version}"
     implementation 'com.amazonaws:aws-encryption-sdk-java:2.4.1'
     implementation 'org.bouncycastle:bcprov-jdk18on:1.77'
+    api "org.apache.httpcomponents.core5:httpcore5:5.2.2"
 
     // ZipArchive dependencies used for integration tests
     zipArchive group: 'org.opensearch.plugin', name:'opensearch-ml-plugin', version: "${opensearch_build}"

src/main/java/org/opensearch/flowframework/indices/FlowFrameworkIndicesHandler.java

@@ -506,8 +506,6 @@ public void updateResourceInStateIndex(
             getResourceByWorkflowStep(workflowStepName),
             resourceId
         );
-        XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent());
-        newResource.toXContent(builder, ToXContent.EMPTY_PARAMS);
 
         // The script to append a new object to the resources_created array
         Script script = new Script(

src/main/java/org/opensearch/flowframework/model/WorkflowState.java

@@ -97,7 +97,7 @@ public WorkflowState(
     private WorkflowState() {}
 
     /**
-     * Instatiates a new WorkflowState from an input stream
+     * Instantiates a new WorkflowState from an input stream
      * @param input the input stream to read from
      * @throws IOException if the workflowId cannot be read from the input stream
      */
@@ -108,10 +108,14 @@ public WorkflowState(StreamInput input) throws IOException {
         this.provisioningProgress = input.readOptionalString();
         this.provisionStartTime = input.readOptionalInstant();
         this.provisionEndTime = input.readOptionalInstant();
-        // TODO: fix error: cannot access Response issue when integrating with access control
-        // this.user = input.readBoolean() ? new User(input) : null;
+        this.user = input.readBoolean() ? new User(input) : null;
         this.userOutputs = input.readBoolean() ? input.readMap() : null;
-        this.resourcesCreated = input.readList(ResourceCreated::new);
+
+        int resourceCount = input.readVInt();
+        this.resourcesCreated = new ArrayList<>(resourceCount);
+        for (int r = 0; r < resourceCount; r++) {
+            resourcesCreated.add(new ResourceCreated(input));
+        }
     }
 
     /**
@@ -293,6 +297,7 @@ public void writeTo(StreamOutput output) throws IOException {
         output.writeOptionalInstant(provisionEndTime);
 
         if (user != null) {
+            output.writeBoolean(true);
             user.writeTo(output);
         } else {
             output.writeBoolean(false);
@@ -304,7 +309,11 @@ public void writeTo(StreamOutput output) throws IOException {
         } else {
             output.writeBoolean(false);
         }
-        output.writeList(resourcesCreated);
+
+        output.writeVInt(resourcesCreated.size());
+        for (ResourceCreated resource : resourcesCreated) {
+            resource.writeTo(output);
+        }
     }
 
     /**

src/test/java/org/opensearch/flowframework/model/WorkflowStateTests.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+package org.opensearch.flowframework.model;
+
+import org.opensearch.common.io.stream.BytesStreamOutput;
+import org.opensearch.commons.authuser.User;
+import org.opensearch.core.common.bytes.BytesReference;
+import org.opensearch.core.common.io.stream.BytesStreamInput;
+import org.opensearch.test.OpenSearchTestCase;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class WorkflowStateTests extends OpenSearchTestCase {
+
+    @Override
+    public void setUp() throws Exception {
+        super.setUp();
+    }
+
+    public void testWorkflowState() throws IOException {
+        String workflowId = "id";
+        String error = "error";
+        String state = "state";
+        String provisioningProgress = "progress";
+        Instant provisionStartTime = Instant.now().minusSeconds(2);
+        Instant provisionEndTime = Instant.now();
+        User user = new User("user", Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
+        Map<String, Object> userOutputs = Map.of("foo", Map.of("bar", "baz"));
+        List<ResourceCreated> resourcesCreated = List.of(new ResourceCreated("name", "stepId", "type", "id"));
+
+        WorkflowState wfs = WorkflowState.builder()
+            .workflowId(workflowId)
+            .error(error)
+            .state(state)
+            .provisioningProgress(provisioningProgress)
+            .provisionStartTime(provisionStartTime)
+            .provisionEndTime(provisionEndTime)
+            .user(user)
+            .userOutputs(userOutputs)
+            .resourcesCreated(resourcesCreated)
+            .build();
+
+        assertEquals(workflowId, wfs.getWorkflowId());
+        assertEquals(error, wfs.getError());
+        assertEquals(state, wfs.getState());
+        assertEquals(provisioningProgress, wfs.getProvisioningProgress());
+        assertEquals(provisionStartTime, wfs.getProvisionStartTime());
+        assertEquals(provisionEndTime, wfs.getProvisionEndTime());
+        assertEquals("user", wfs.getUser().getName());
+        assertEquals(1, wfs.userOutputs().size());
+        assertEquals("baz", ((Map<?, ?>) wfs.userOutputs().get("foo")).get("bar"));
+        assertEquals(1, wfs.resourcesCreated().size());
+        ResourceCreated rc = wfs.resourcesCreated().get(0);
+        assertEquals("name", rc.workflowStepName());
+        assertEquals("stepId", rc.workflowStepId());
+        assertEquals("type", rc.resourceType());
+        assertEquals("id", rc.resourceId());
+
+        try (BytesStreamOutput out = new BytesStreamOutput()) {
+            wfs.writeTo(out);
+            try (BytesStreamInput in = new BytesStreamInput(BytesReference.toBytes(out.bytes()))) {
+                wfs = new WorkflowState(in);
+
+                assertEquals(workflowId, wfs.getWorkflowId());
+                assertEquals(error, wfs.getError());
+                assertEquals(state, wfs.getState());
+                assertEquals(provisioningProgress, wfs.getProvisioningProgress());
+                assertEquals(provisionStartTime, wfs.getProvisionStartTime());
+                assertEquals(provisionEndTime, wfs.getProvisionEndTime());
+                assertEquals("user", wfs.getUser().getName());
+                assertEquals(1, wfs.userOutputs().size());
+                assertEquals("baz", ((Map<?, ?>) wfs.userOutputs().get("foo")).get("bar"));
+                assertEquals(1, wfs.resourcesCreated().size());
+                rc = wfs.resourcesCreated().get(0);
+                assertEquals("name", rc.workflowStepName());
+                assertEquals("stepId", rc.workflowStepId());
+                assertEquals("type", rc.resourceType());
+                assertEquals("id", rc.resourceId());
+            }
+        }
+    }
+
+}