fix: Scheduling synchronized block on bounded elastic threadpool instead of main event threadpool (#32343)

…

## Description
> [!TIP]  
> _Add a TL;DR when the description is longer than 500 words or
extremely technical (helps the content team)._
>
> _Please also include relevant motivation and context. List any
dependencies that are required for this change. Add links to Notion,
Figma or any other documents that might be relevant to the PR._

Fixes #`Issue Number`  
_or_  
Fixes `Issue URL`
> [!WARNING]  
> _If no issue exists, please create an issue first, and check with the
maintainers if the issue is valid._

## Automation

/ok-to-test tags="@tag.Sanity"

### 🔍 Cypress test results
<!-- This is an auto-generated comment: Cypress test results  -->
> [!IMPORTANT]  
> Workflow run:
<https://github.com/appsmithorg/appsmith/actions/runs/8523943321>
> Commit: `03dc817039ee24159438e613320550620353a2b7`
> Cypress dashboard url: <a
href="https://internal.appsmith.com/app/cypress-dashboard/rundetails-65890b3c81d7400d08fa9ee5?branch=master&workflowId=8523943321&attempt=1"
target="_blank">Click here!</a>
> All cypress tests have passed 🎉🎉🎉

<!-- end of auto-generated comment: Cypress test results  -->



<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit


- **Refactor**
- Improved backend scheduling for datasource operations to enhance
performance without blocking the main thread.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: trishaanand

app/server/appsmith-server/src/main/java/com/appsmith/server/services/ce/DatasourceContextServiceCEImpl.java

@@ -24,6 +24,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import java.time.Instant;
 import java.util.Map;
@@ -103,7 +104,8 @@ public Mono<? extends DatasourceContext<?>> getCachedDatasourceContextMono(
                         // Basically remove entry from both cache maps
                         pluginExecutor.datasourceDestroy(connection);
                     } catch (Exception e) {
-                        log.info("Error destroying stale datasource connection", e);
+                        log.info(
+                                Thread.currentThread().getName() + ": Error destroying stale datasource connection", e);
                     }
                 }
                 datasourceContextMonoMap.remove(datasourceContextIdentifier);
@@ -117,7 +119,8 @@ public Mono<? extends DatasourceContext<?>> getCachedDatasourceContextMono(
              */
             if (datasourceContextIdentifier.getDatasourceId() != null
                     && datasourceContextMonoMap.get(datasourceContextIdentifier) != null) {
-                log.debug("Cached resource context mono exists. Returning the same.");
+                log.debug(Thread.currentThread().getName()
+                        + ": Cached resource context mono exists. Returning the same.");
                 return datasourceContextMonoMap.get(datasourceContextIdentifier);
             }
 
@@ -182,11 +185,11 @@ public Mono<Object> updateDatasourceAndSetAuthentication(Object connection, Data
 
     protected Mono<DatasourceContext<?>> createNewDatasourceContext(
             DatasourceStorage datasourceStorage, DatasourceContextIdentifier datasourceContextIdentifier) {
-        log.debug("Datasource context doesn't exist. Creating connection.");
+        log.debug(Thread.currentThread().getName() + ": Datasource context doesn't exist. Creating connection.");
         Mono<Plugin> pluginMono =
                 pluginService.findById(datasourceStorage.getPluginId()).cache();
 
-        return pluginMono
+        return (Mono<DatasourceContext<?>>) pluginMono
                 .zipWith(pluginExecutorHelper.getPluginExecutor(pluginMono))
                 .flatMap(tuple2 -> {
                     Plugin plugin = tuple2.getT1();
@@ -214,7 +217,9 @@ protected Mono<DatasourceContext<?>> createNewDatasourceContext(
 
                     return getCachedDatasourceContextMono(
                             datasourceStorage, plugin, pluginExecutor, monitor, datasourceContextIdentifier);
-                });
+                })
+                // Scheduling on bounded elastic to avoid blocking the main thread
+                .subscribeOn(Schedulers.boundedElastic());
     }
 
     public boolean getIsStale(