AsyncRestExecutor implements RestExecutor { @Override public void execute(Client client, Map<String, String> params, QueryAction queryAction, RestChannel channel) throws Exception { if (isBlockingAction(queryAction) && isRunningInTransportThread()) { if (LOG.isDebugEnabled()) { LOG.debug("[{}] Async blocking query action [{}] for executor [{}] in current thread [{}]", LogUtils.getRequestId(), name(executor), name(queryAction), Thread.currentThread().getName()); } async(client, params, queryAction, channel); } else { if (LOG.isDebugEnabled()) { LOG.debug("[{}] Continue running query action [{}] for executor [{}] in current thread [{}]", LogUtils.getRequestId(), name(executor), name(queryAction), Thread.currentThread().getName()); } doExecuteWithTimeMeasured(client, params, queryAction, channel); } } AsyncRestExecutor(RestExecutor executor);   AsyncRestExecutor(RestExecutor executor, Predicate<QueryAction> isBlocking); @Override void execute(Client client, Map<String, String> params, QueryAction queryAction, RestChannel channel); @Override String execute(Client client, Map<String, String> params, QueryAction queryAction); static final String SQL_WORKER_THREAD_POOL_NAME; }
@Test public void executeBlockingQuery() throws Exception { Thread.currentThread().setName(TRANSPORT_WORKER_THREAD_NAME_PREFIX); execute(); verifyRunInWorkerThread(); }
@Test public void executeBlockingQueryButNotInTransport() throws Exception { execute(); verifyRunInCurrentThread(); }
@Test public void executeNonBlockingQuery() throws Exception { execute(anyAction -> NON_BLOCKING); verifyRunInCurrentThread(); }