Run update/delete request in a new thread (#337)

This is not to block transport thread Signed-off-by: Heemin Kim <[email protected]>
opensearch-project · Jun 16, 2023 · 63e3a47 · 63e3a47
1 parent e082a6f
commit 63e3a47
Show file tree

Hide file tree

Showing 4 changed files with 41 additions and 16 deletions.
diff --git a/src/main/java/org/opensearch/geospatial/ip2geo/action/DeleteDatasourceTransportAction.java b/src/main/java/org/opensearch/geospatial/ip2geo/action/DeleteDatasourceTransportAction.java
@@ -26,6 +26,7 @@
 import org.opensearch.geospatial.ip2geo.jobscheduler.Datasource;
 import org.opensearch.ingest.IngestService;
 import org.opensearch.tasks.Task;
+import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportService;
 
 /**
@@ -39,6 +40,7 @@ public class DeleteDatasourceTransportAction extends HandledTransportAction<Dele
     private final DatasourceFacade datasourceFacade;
     private final GeoIpDataFacade geoIpDataFacade;
     private final Ip2GeoProcessorFacade ip2GeoProcessorFacade;
+    private final ThreadPool threadPool;
 
     /**
      * Constructor
@@ -56,14 +58,16 @@ public DeleteDatasourceTransportAction(
         final IngestService ingestService,
         final DatasourceFacade datasourceFacade,
         final GeoIpDataFacade geoIpDataFacade,
-        final Ip2GeoProcessorFacade ip2GeoProcessorFacade
+        final Ip2GeoProcessorFacade ip2GeoProcessorFacade,
+        final ThreadPool threadPool
     ) {
         super(DeleteDatasourceAction.NAME, transportService, actionFilters, DeleteDatasourceRequest::new);
         this.lockService = lockService;
         this.ingestService = ingestService;
         this.datasourceFacade = datasourceFacade;
         this.geoIpDataFacade = geoIpDataFacade;
         this.ip2GeoProcessorFacade = ip2GeoProcessorFacade;
+        this.threadPool = threadPool;
     }
 
     /**
@@ -83,9 +87,17 @@ protected void doExecute(final Task task, final DeleteDatasourceRequest request,
                 return;
             }
             try {
-                deleteDatasource(request.getName());
-                lockService.releaseLock(lock);
-                listener.onResponse(new AcknowledgedResponse(true));
+                // TODO: makes every sub-methods as async call to avoid using a thread in generic pool
+                threadPool.generic().submit(() -> {
+                    try {
+                        deleteDatasource(request.getName());
+                        lockService.releaseLock(lock);
+                        listener.onResponse(new AcknowledgedResponse(true));
+                    } catch (Exception e) {
+                        lockService.releaseLock(lock);
+                        listener.onFailure(e);
+                    }
+                });
             } catch (Exception e) {
                 lockService.releaseLock(lock);
                 listener.onFailure(e);

diff --git a/src/main/java/org/opensearch/geospatial/ip2geo/action/UpdateDatasourceTransportAction.java b/src/main/java/org/opensearch/geospatial/ip2geo/action/UpdateDatasourceTransportAction.java
@@ -31,6 +31,7 @@
 import org.opensearch.geospatial.ip2geo.jobscheduler.DatasourceUpdateService;
 import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule;
 import org.opensearch.tasks.Task;
+import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportService;
 
 /**
@@ -42,6 +43,7 @@ public class UpdateDatasourceTransportAction extends HandledTransportAction<Upda
     private final Ip2GeoLockService lockService;
     private final DatasourceFacade datasourceFacade;
     private final DatasourceUpdateService datasourceUpdateService;
+    private final ThreadPool threadPool;
 
     /**
      * Constructor
@@ -58,12 +60,14 @@ public UpdateDatasourceTransportAction(
         final ActionFilters actionFilters,
         final Ip2GeoLockService lockService,
         final DatasourceFacade datasourceFacade,
-        final DatasourceUpdateService datasourceUpdateService
+        final DatasourceUpdateService datasourceUpdateService,
+        final ThreadPool threadPool
     ) {
         super(UpdateDatasourceAction.NAME, transportService, actionFilters, UpdateDatasourceRequest::new);
         this.lockService = lockService;
         this.datasourceUpdateService = datasourceUpdateService;
         this.datasourceFacade = datasourceFacade;
+        this.threadPool = threadPool;
     }
 
     /**
@@ -82,16 +86,23 @@ protected void doExecute(final Task task, final UpdateDatasourceRequest request,
                 );
                 return;
             }
-
             try {
-                Datasource datasource = datasourceFacade.getDatasource(request.getName());
-                if (datasource == null) {
-                    throw new ResourceNotFoundException("no such datasource exist");
-                }
-                validate(request, datasource);
-                updateIfChanged(request, datasource);
-                lockService.releaseLock(lock);
-                listener.onResponse(new AcknowledgedResponse(true));
+                // TODO: makes every sub-methods as async call to avoid using a thread in generic pool
+                threadPool.generic().submit(() -> {
+                    try {
+                        Datasource datasource = datasourceFacade.getDatasource(request.getName());
+                        if (datasource == null) {
+                            throw new ResourceNotFoundException("no such datasource exist");
+                        }
+                        validate(request, datasource);
+                        updateIfChanged(request, datasource);
+                        lockService.releaseLock(lock);
+                        listener.onResponse(new AcknowledgedResponse(true));
+                    } catch (Exception e) {
+                        lockService.releaseLock(lock);
+                        listener.onFailure(e);
+                    }
+                });
             } catch (Exception e) {
                 lockService.releaseLock(lock);
                 listener.onFailure(e);

diff --git a/...st/java/org/opensearch/geospatial/ip2geo/action/DeleteDatasourceTransportActionTests.java b/...st/java/org/opensearch/geospatial/ip2geo/action/DeleteDatasourceTransportActionTests.java
@@ -48,7 +48,8 @@ public void init() {
             ingestService,
             datasourceFacade,
             geoIpDataFacade,
-            ip2GeoProcessorFacade
+            ip2GeoProcessorFacade,
+            threadPool
         );
     }
 

diff --git a/...st/java/org/opensearch/geospatial/ip2geo/action/UpdateDatasourceTransportActionTests.java b/...st/java/org/opensearch/geospatial/ip2geo/action/UpdateDatasourceTransportActionTests.java
@@ -45,7 +45,8 @@ public void init() {
             actionFilters,
             ip2GeoLockService,
             datasourceFacade,
-            datasourceUpdateService
+            datasourceUpdateService,
+            threadPool
         );
     }