private void pollWithExecutorPool(long timeoutMs, List<NetworkSend> sends) throws IOException {
    List<Future<NetworkSend>> completedSendsFutures = new ArrayList<>();
    List<Future<NetworkReceive>> completedReceivesFutures = new ArrayList<>();
    List<Future<SelectionKey>> completedPrepareFutures = new ArrayList<>();
    Set<SelectionKey> processingKeySet = new HashSet<>();
    clear();

    // register for write interest on any new sends
    if (sends != null) {
      for (NetworkSend networkSend : sends) {
        send(networkSend);
      }
    }

    // check ready keys
    long startSelect = time.milliseconds();
    int readyKeys = select(timeoutMs);
    this.metrics.selectorSelectCount.inc();

    if (readyKeys > 0) {
      long endSelect = time.milliseconds();
      this.metrics.selectorSelectTime.update(endSelect - startSelect);
      Set<SelectionKey> keys = nioSelector.selectedKeys();
      metrics.selectorReadyKeyCount.inc(keys.size());
      Iterator<SelectionKey> iter = keys.iterator();
      while (iter.hasNext()) {
        if (networkConfig.selectorMaxKeyToProcess != -1
            && processingKeySet.size() >= networkConfig.selectorMaxKeyToProcess) {
          break;
        }
        SelectionKey key = iter.next();
        iter.remove();

        Transmission transmission = getTransmission(key);
        try {
          if (key.isConnectable()) {
            transmission.finishConnect();
            if (transmission.ready()) {
              connected.add(transmission.getConnectionId());
              metrics.selectorConnectionCreated.inc();
            } else {
              unreadyConnections.add(transmission.getConnectionId());
            }
          }

          /* if channel is not ready, finish prepare */
          if (transmission.isConnected() && !transmission.ready()) {
            metrics.selectorPrepareKeyCount.inc();
            completedPrepareFutures.add(executorPool.submit(() -> prepare(key, transmission)));
            processingKeySet.add(key);
          } else if (key.isReadable() && transmission.ready()) {
            metrics.selectorReadKeyCount.inc();
            completedReceivesFutures.add(executorPool.submit(() -> read(key, transmission)));
            processingKeySet.add(key);
          } else if (key.isWritable() && transmission.ready()) {
            metrics.selectorWriteKeyCount.inc();
            completedSendsFutures.add(executorPool.submit(() -> write(key, transmission)));
            processingKeySet.add(key);
          } else if (!key.isValid()) {
            close(key);
          }
        } catch (IOException e) {
          // handles IOException from transmission.finishConnect()
          handleReadWriteIOException(e, key);
          close(key);
        } catch (Exception e) {
          close(key);
          metrics.selectorKeyOperationErrorCount.inc();
          logger.error("closing key on exception remote host {}", channel(key).socket().getRemoteSocketAddress(), e);
        }
      }

      for (Future<SelectionKey> future : completedPrepareFutures) {
        try {
          SelectionKey returnKey = future.get();
          if (returnKey != null) {
            processingKeySet.remove(returnKey);
          }
        } catch (InterruptedException | ExecutionException e) {
          logger.error("Hit Unexpected exception on selector prepare, ", e);
        }
      }

      for (Future<NetworkReceive> future : completedReceivesFutures) {
        try {
          NetworkReceive networkReceive = future.get();
          if (networkReceive != null) {
            processingKeySet.remove(keyForId(networkReceive.getConnectionId()));
            if (networkReceive.getReceivedBytes().isReadComplete()) {
              this.completedReceives.add(networkReceive);
            }
          }
        } catch (InterruptedException | ExecutionException e) {
          logger.error("Hit Unexpected exception on selector read, ", e);
        }
      }

      for (Future<NetworkSend> future : completedSendsFutures) {
        try {
          NetworkSend networkSend = future.get();
          if (networkSend != null) {
            processingKeySet.remove(keyForId(networkSend.getConnectionId()));
            if (networkSend.getPayload().isSendComplete()) {
              this.completedSends.add(networkSend);
            }
          }
        } catch (ExecutionException | InterruptedException e) {
          logger.error("Hit Unexpected exception on selector write, ", e);
        }
      }
      // Keys hit exception should be closed.
      for (SelectionKey keyWithError : processingKeySet) {
        close(keyWithError);
      }
      checkUnreadyConnectionsStatus();
      long selectorIOTime = time.milliseconds() - endSelect;
      if (selectorIOTime >= 50) {
        // Some selectors may not have enough work to do, so counting their time is misleading.
        metrics.selectorIOCount.inc();
        metrics.selectorIOTime.update(time.milliseconds() - endSelect);
      }
    }
    disconnected.addAll(closedConnections);
    closedConnections.clear();
  }
Use {@link ExecutorService} to process read/write events.
ambry-network/src/main/java/com/github/ambry/network/Selector.java