Fix dynamic index refresh to target written indices instead of global refresh

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

CHANGELOG.md

@@ -12,6 +12,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ### Removed
 
 ### Fixed
+- Fixed global refresh when using dynamic index patterns ([#324](https://github.com/opensearch-project/opensearch-hadoop/issues/324))
 - Fixed build failures when downloading Apache project dependencies (Hadoop, Hive, Spark) ([#595](https://github.com/opensearch-project/opensearch-hadoop/pull/595))
 
 ### Security

mr/src/main/java/org/opensearch/hadoop/rest/RestClient.java

@@ -299,6 +299,10 @@ public void refresh(Resource resource) {
         execute(POST, resource.refresh());
     }
 
+    public void refreshIndex(String index) {
+        execute(POST, index + "/_refresh");
+    }
+
     public List<List<Map<String, Object>>> targetShards(String index, String routing) {
         List<List<Map<String, Object>>> shardsJson = null;
 

mr/src/main/java/org/opensearch/hadoop/rest/bulk/BulkProcessor.java

@@ -32,8 +32,10 @@
 import java.io.Closeable;
 import java.util.ArrayList;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -90,6 +92,7 @@ public class BulkProcessor implements Closeable, StatsAware {
     private boolean executedBulkWrite = false;
     private boolean hadWriteErrors = false;
     private boolean requiresRefreshAfterBulk = false;
+    private final Set<String> writtenIndices = new LinkedHashSet<>();
 
     // Bulk write error handlers.
     private List<IBulkWriteErrorHandler> documentBulkErrorHandlers;
@@ -275,6 +278,10 @@ public BulkResponse tryFlush() {
 
                             if (error == null){
                                 // Write operation for this entry succeeded
+                                String idx = (String) values.get("_index");
+                                if (idx != null) {
+                                    writtenIndices.add(idx);
+                                }
                                 stats.bytesAccepted += data.length(trackingBytesPosition);
                                 stats.docsAccepted += 1;
                                 docsSent += 1;
@@ -574,12 +581,13 @@ public void close() {
                 }
             }
 
-            if (requiresRefreshAfterBulk && executedBulkWrite) {
-                // refresh batch
-                restClient.refresh(resource);
+            if (requiresRefreshAfterBulk && executedBulkWrite && !writtenIndices.isEmpty()) {
+                for (String idx : writtenIndices) {
+                    restClient.refreshIndex(idx);
+                }
 
                 if (LOG.isDebugEnabled()) {
-                    LOG.debug(String.format("Refreshing index [%s]", resource));
+                    LOG.debug(String.format("Refreshing index %s", writtenIndices));
                 }
             }
         } finally {

mr/src/test/java/org/opensearch/hadoop/rest/bulk/BulkOutputGenerator.java

@@ -43,6 +43,8 @@ public interface BulkOutputGenerator {
 
     public BulkOutputGenerator addSuccess(String operation, int status);
 
+    public BulkOutputGenerator addSuccess(String operation, int status, String index);
+
     public BulkOutputGenerator addFailure(String operation, int status, String type, String errorMessage);
 
     public BulkOutputGenerator addRejection(String operation);

mr/src/test/java/org/opensearch/hadoop/rest/bulk/BulkProcessorTest.java

@@ -967,4 +967,80 @@ private static byte[] copyDocumentBytes(BulkWriteFailure entry) throws IOExcepti
         assertEquals(-1, inputStream.read());
         return copyDoc;
     }
+
+    @Test
+    public void testBulk10_RefreshSingleIndex() throws Exception {
+        RestClient.BulkActionResponse response = generator.setInfo(resource, 56)
+                .addSuccess("index", 201)
+                .addSuccess("index", 201)
+                .addSuccess("index", 201)
+                .generate();
+
+        RestClient mockClient = mockClientResponses(response);
+        BulkProcessor processor = new BulkProcessor(mockClient, resource, testSettings);
+
+        BytesRef data = new BytesRef();
+        data.add(renderEntry("A"));
+        processor.add(data);
+        data.reset();
+        data.add(renderEntry("B"));
+        processor.add(data);
+        data.reset();
+        data.add(renderEntry("C"));
+        processor.add(data);
+
+        processor.close();
+
+        Mockito.verify(mockClient).refreshIndex("foo");
+        Mockito.verify(mockClient, Mockito.never()).refresh(Mockito.any(Resource.class));
+    }
+
+    @Test
+    public void testBulk10_RefreshMultipleDynamicIndices() throws Exception {
+        RestClient.BulkActionResponse response = generator.setInfo(resource, 56)
+                .addSuccess("index", 201, "logs-2026-01")
+                .addSuccess("index", 201, "logs-2026-02")
+                .addSuccess("index", 201, "logs-2026-01")
+                .addSuccess("index", 201, "logs-2026-03")
+                .addSuccess("index", 201, "logs-2026-02")
+                .generate();
+
+        RestClient mockClient = mockClientResponses(response);
+        BulkProcessor processor = new BulkProcessor(mockClient, resource, testSettings);
+
+        processData(processor);
+
+        processor.close();
+
+        Mockito.verify(mockClient).refreshIndex("logs-2026-01");
+        Mockito.verify(mockClient).refreshIndex("logs-2026-02");
+        Mockito.verify(mockClient).refreshIndex("logs-2026-03");
+        Mockito.verify(mockClient, Mockito.times(3)).refreshIndex(Mockito.anyString());
+        Mockito.verify(mockClient, Mockito.never()).refresh(Mockito.any(Resource.class));
+    }
+
+    @Test
+    public void testBulk10_NoRefreshWhenDisabled() throws Exception {
+        testSettings.setProperty(ConfigurationOptions.OPENSEARCH_BATCH_WRITE_REFRESH, "false");
+
+        RestClient.BulkActionResponse response = generator.setInfo(resource, 56)
+                .addSuccess("index", 201)
+                .addSuccess("index", 201)
+                .generate();
+
+        RestClient mockClient = mockClientResponses(response);
+        BulkProcessor processor = new BulkProcessor(mockClient, resource, testSettings);
+
+        BytesRef data = new BytesRef();
+        data.add(renderEntry("A"));
+        processor.add(data);
+        data.reset();
+        data.add(renderEntry("B"));
+        processor.add(data);
+
+        processor.close();
+
+        Mockito.verify(mockClient, Mockito.never()).refreshIndex(Mockito.anyString());
+        Mockito.verify(mockClient, Mockito.never()).refresh(Mockito.any(Resource.class));
+    }
 }

mr/src/test/java/org/opensearch/hadoop/rest/bulk/bwc/BulkOutputGeneratorBase.java

@@ -115,10 +115,15 @@ public BulkOutputGenerator setInfo(Resource resource, long took) {
 
     @Override
     public BulkOutputGenerator addSuccess(String operation, int status) {
+        return addSuccess(operation, status, resource.index());
+    }
+
+    @Override
+    public BulkOutputGenerator addSuccess(String operation, int status, String index) {
         Assert.notNull(resource);
         items.add(getSuccess()
                 .replace(OP, operation)
-                .replace(IDX, resource.index())
+                .replace(IDX, index)
                 .replace(TYPE, resource.type())
                 .replace(ID, UUID.randomUUID().toString())
                 .replace(VER, "1")