BulkRequest: Make sure that indices Set is properly initialized when being deserialized from TransportRequest stream (#20132)

* BulkRequest: Make sure that indices Set is properly initialized when being deserialized from TransportRequest stream

Signed-off-by: Nils Bandener <[email protected]>

* fix

Signed-off-by: Nils Bandener <[email protected]>

* simplified BytesStreamOutput construction

Signed-off-by: Nils Bandener <[email protected]>

---------

Signed-off-by: Nils Bandener <[email protected]>

Author: nibix

CHANGELOG.md

@@ -93,6 +93,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix node bootstrap error when enable stream transport and remote cluster state ([#19948](https://github.com/opensearch-project/OpenSearch/pull/19948))
 - Fix deletion failure/error of unused index template; case when an index template matches a data stream but has a lower priority. ([#20102](https://github.com/opensearch-project/OpenSearch/pull/20102))
 - Fix toBuilder method in EngineConfig to include mergedSegmentTransferTracker([20105](https://github.com/opensearch-project/OpenSearch/pull/20105))
+- Fixed handling of property index in BulkRequest during deserialization ([#20132](https://github.com/opensearch-project/OpenSearch/pull/20132))
 
 ### Dependencies
 - Bump Apache Lucene from 10.3.1 to 10.3.2 ([#20026](https://github.com/opensearch-project/OpenSearch/pull/20026))

server/src/main/java/org/opensearch/action/bulk/BulkRequest.java

@@ -111,6 +111,7 @@ public BulkRequest(StreamInput in) throws IOException {
         if (in.getVersion().onOrAfter(Version.V_2_14_0) && in.getVersion().before(Version.V_3_0_0)) {
             in.readInt(); // formerly batch_size
         }
+        requests.stream().map(DocWriteRequest::index).forEach(indices::add);
     }
 
     public BulkRequest(@Nullable String globalIndex) {

server/src/test/java/org/opensearch/action/bulk/TransportBulkActionTests.java

@@ -53,6 +53,7 @@
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodeRole;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.io.stream.BytesStreamOutput;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
@@ -77,6 +78,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
@@ -379,6 +381,15 @@ public void testRejectionAfterCreateIndexIsPropagated() throws Exception {
         }
     }
 
+    public void testSerializationDeserialization() throws Exception {
+        BulkRequest bulkRequest = new BulkRequest().add(new IndexRequest("index").id("id").source(Collections.emptyMap()));
+        BytesStreamOutput out = new BytesStreamOutput();
+        out.setVersion(Version.V_3_4_0);
+        bulkRequest.writeTo(out);
+        BulkRequest deserializedRequest = new BulkRequest(out.bytes().streamInput());
+        assertEquals(Set.of("index"), deserializedRequest.getIndices());
+    }
+
     private BulkRequest buildBulkRequest(List<String> indices) {
         BulkRequest request = new BulkRequest();
         for (String index : indices) {