run sendRequest in an executor (#341)

* run sendRequest in an executor

Author: sji01

.gitignore

@@ -15,3 +15,4 @@ target
 /.gradle
 .settings
 subprojects/parseq-lambda-names/bin/
+subprojects/parseq-tracevis/npm-debug.log

CHANGELOG.md

@@ -1,3 +1,7 @@
+v5.1.16
+------
+* Add the support of offloading sendRequest call to an executor
+
 v5.1.15
 ------
 * Fix stack overflow error in TaskDescriptor when lambda is used

gradle.properties

@@ -1,3 +1,3 @@
-version=5.1.15
+version=5.1.16
 group=com.linkedin.parseq
 org.gradle.parallel=true

subprojects/parseq-restli-client/src/main/java/com/linkedin/restli/client/DirectExecutor.java

@@ -0,0 +1,21 @@
+package com.linkedin.restli.client;
+
+import java.util.concurrent.Executor;
+
+
+/**
+ * A simple executor implementation that executes the task immediately on the calling thread
+ */
+class DirectExecutor implements Executor {
+
+  private static final DirectExecutor INSTANCE = new DirectExecutor();
+
+  static DirectExecutor getInstance() {
+    return INSTANCE;
+  }
+
+  @Override
+  public void execute(Runnable command) {
+    command.run();
+  }
+}

subprojects/parseq-restli-client/src/main/java/com/linkedin/restli/client/ParSeqRestClient.java

@@ -35,6 +35,7 @@
 import com.linkedin.restli.client.metrics.Metrics;
 import com.linkedin.restli.common.OperationNameGenerator;
 import java.util.Optional;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import org.slf4j.Logger;
@@ -69,16 +70,24 @@ public class ParSeqRestClient extends BatchingStrategy<RequestGroup, RestRequest
   private final RequestConfigProvider _requestConfigProvider;
   private final boolean _d2RequestTimeoutEnabled;
   private final Function<Request<?>, RequestContext> _requestContextProvider;
+  private final Executor _executor;
 
   ParSeqRestClient(final Client client, final RequestConfigProvider requestConfigProvider,
-      Function<Request<?>, RequestContext> requestContextProvider, final boolean d2RequestTimeoutEnabled) {
+      Function<Request<?>, RequestContext> requestContextProvider, final boolean d2RequestTimeoutEnabled,
+      Executor executor) {
     ArgumentUtil.requireNotNull(client, "client");
     ArgumentUtil.requireNotNull(requestConfigProvider, "requestConfigProvider");
     ArgumentUtil.requireNotNull(requestContextProvider, "requestContextProvider");
     _client = client;
     _requestConfigProvider = requestConfigProvider;
     _requestContextProvider = requestContextProvider;
     _d2RequestTimeoutEnabled = d2RequestTimeoutEnabled;
+    _executor = executor;
+  }
+
+  ParSeqRestClient(final Client client, final RequestConfigProvider requestConfigProvider,
+      Function<Request<?>, RequestContext> requestContextProvider, final boolean d2RequestTimeoutEnabled) {
+    this(client, requestConfigProvider, requestContextProvider, d2RequestTimeoutEnabled, DirectExecutor.getInstance());
   }
 
   /**
@@ -93,6 +102,7 @@ public ParSeqRestClient(final Client client) {
     _requestConfigProvider = RequestConfigProvider.build(new ParSeqRestliClientConfigBuilder().build(), () -> Optional.empty());
     _requestContextProvider = request -> new RequestContext();
     _d2RequestTimeoutEnabled = false;
+    _executor = DirectExecutor.getInstance();
   }
 
   /**
@@ -107,6 +117,7 @@ public ParSeqRestClient(final RestClient client) {
     _requestConfigProvider = RequestConfigProvider.build(new ParSeqRestliClientConfigBuilder().build(), () -> Optional.empty());
     _requestContextProvider = request -> new RequestContext();
     _d2RequestTimeoutEnabled = false;
+    _executor = DirectExecutor.getInstance();
   }
 
   @Override
@@ -119,11 +130,13 @@ public <T> Promise<Response<T>> sendRequest(final Request<T> request) {
   @Deprecated
   public <T> Promise<Response<T>> sendRequest(final Request<T> request, final RequestContext requestContext) {
     final SettablePromise<Response<T>> promise = Promises.settable();
-    try {
-      _client.sendRequest(request, requestContext, new PromiseCallbackAdapter<T>(promise));
-    } catch (Throwable e) {
-      promise.fail(e);
-    }
+    _executor.execute(() -> {
+      try {
+        _client.sendRequest(request, requestContext, new PromiseCallbackAdapter<T>(promise));
+      } catch (Throwable e) {
+        promise.fail(e);
+      }
+    });
     return promise;
   }
 

subprojects/parseq-restli-client/src/main/java/com/linkedin/restli/client/ParSeqRestliClientBuilder.java

@@ -3,6 +3,7 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.Executor;
 import java.util.function.Function;
 
 import org.slf4j.Logger;
@@ -35,6 +36,7 @@ public class ParSeqRestliClientBuilder {
   private BatchingSupport _batchingSupport;
   private InboundRequestContextFinder _inboundRequestContextFinder;
   private Function<Request<?>, RequestContext> _requestContextProvider;
+  private Executor _executor = DirectExecutor.getInstance();
 
   /**
    * This method may throw RuntimeException e.g. when there is a problem with configuration.
@@ -65,7 +67,7 @@ public ParSeqRestClient build() {
         request -> new RequestContext() :
         _requestContextProvider;
 
-    ParSeqRestClient parseqClient = new ParSeqRestClient(_client, configProvider, requestContextProvider, _d2RequestTimeoutEnabled);
+    ParSeqRestClient parseqClient = new ParSeqRestClient(_client, configProvider, requestContextProvider, _d2RequestTimeoutEnabled, _executor);
     if (_batchingSupport != null) {
       LOGGER.debug("Found batching support");
       _batchingSupport.registerStrategy(parseqClient);
@@ -160,4 +162,9 @@ public ParSeqRestliClientBuilder setD2RequestTimeoutEnabled(boolean enabled) {
     _d2RequestTimeoutEnabled = enabled;
     return this;
   }
+
+  public ParSeqRestliClientBuilder setExecutor(Executor executor) {
+    _executor = executor;
+    return this;
+  }
 }

subprojects/parseq/src/main/java/com/linkedin/parseq/FusionTask.java

@@ -99,11 +99,11 @@ private ShallowTraceBuilder getEffectiveShallowTraceBuilder(final FusionTraceCon
   }
 
   private void addRelationships(final FusionTraceContext traceContext) {
-    final ShallowTraceBuilder effectoveShallowTraceBuilder = getEffectiveShallowTraceBuilder(traceContext);
+    final ShallowTraceBuilder effectiveShallowTraceBuilder = getEffectiveShallowTraceBuilder(traceContext);
     TraceBuilder builder = getTraceBuilder();
-    builder.addRelationship(Relationship.PARENT_OF, traceContext.getParent().getShallowTraceBuilder(), effectoveShallowTraceBuilder);
+    builder.addRelationship(Relationship.PARENT_OF, traceContext.getParent().getShallowTraceBuilder(), effectiveShallowTraceBuilder);
     if (_predecessorShallowTraceBuilder != null) {
-      builder.addRelationship(Relationship.SUCCESSOR_OF, effectoveShallowTraceBuilder, _predecessorShallowTraceBuilder);
+      builder.addRelationship(Relationship.SUCCESSOR_OF, effectiveShallowTraceBuilder, _predecessorShallowTraceBuilder);
     }
   }