docs: boolean queue(
              ProcedureInvocation invocation,
              ProcedureCallback cb,
              final boolean ignoreBackpressure, final long nowNanos, final long timeoutNanos)
              throws NoConnectionsException {
          assert(invocation != null);
          assert(cb != null);
  
          NodeConnection cxn = null;
          boolean backpressure = true;
  
          /*
           * Synchronization is necessary to ensure that m_connections is not modified
           * as well as to ensure that backpressure is reported correctly
           */
          synchronized (this) {
              final int totalConnections = m_connections.size();
  
              if (totalConnections == 0) {
                  throw new NoConnectionsException("No connections.");
              }
  
              /*
               * Check if the master for the partition is known. No back pressure check to ensure correct
               * routing, but backpressure will be managed anyways. This is where we guess partition based on client
               * affinity and known topology (hashinator initialized).
               */
              if (m_useClientAffinity && (m_hashinator != null)) {
                  final ImmutableSortedMap<String, Procedure> procedures = m_procedureInfo.get();
                  Procedure procedureInfo = null;
                  if (procedures != null) {
                      procedureInfo = procedures.get(invocation.getProcName());
                  }
                  Integer hashedPartition = -1;
  
                  if (procedureInfo != null) {
                      hashedPartition = Constants.MP_INIT_PID;
                      if (( ! procedureInfo.multiPart) &&
                          // User may have passed too few parameters to allow dispatching.
                          // Avoid an indexing error here to fall through to the proper ProcCallException.
                              (procedureInfo.partitionParameter < invocation.getPassedParamCount())) {
                          hashedPartition = m_hashinator.getHashedPartitionForParameter(
                                  procedureInfo.partitionParameterType,
                                  invocation.getPartitionParamValue(procedureInfo.partitionParameter));
                      }
                      /*
                       * If the procedure is read only and single part and the user wants it, load balance across replicas
                       * This is probably slower for SAFE consistency.
                       */
                      if (!procedureInfo.multiPart && procedureInfo.readOnly && m_sendReadsToReplicasBytDefaultIfCAEnabled) {
                          NodeConnection partitionReplicas[] = m_partitionReplicas.get(hashedPartition);
                          if (partitionReplicas != null && partitionReplicas.length > 0) {
                              cxn = partitionReplicas[ThreadLocalRandom.current().nextInt(partitionReplicas.length)];
                              if (cxn.hadBackPressure()) {
                                  //See if there is one without backpressure, make sure it's still connected
                                  for (NodeConnection nc : partitionReplicas) {
                                      if (!nc.hadBackPressure() && nc.m_isConnected) {
                                          cxn = nc;
                                          break;
                                      }
                                  }
                              }
                              if (!cxn.hadBackPressure() || ignoreBackpressure) {
                                  backpressure = false;
                              }
                          }
                      } else {
                          /*
                           * For writes or SAFE reads, this is the best way to go
                           */
                          cxn = m_partitionMasters.get(hashedPartition);
                          if (cxn != null && !cxn.hadBackPressure() || ignoreBackpressure) {
                              backpressure = false;
                          }
                      }
                  }
                  if (cxn != null && !cxn.m_isConnected) {
                      // Would be nice to log something here
                      // Client affinity picked a connection that was actually disconnected.  Reset to null
                      // and let the round-robin choice pick a connection
                      cxn = null;
                  }
                  ClientAffinityStats stats = m_clientAffinityStats.get(hashedPartition);
                  if (stats == null) {
                      stats = new ClientAffinityStats(hashedPartition, 0, 0, 0, 0);
                      m_clientAffinityStats.put(hashedPartition, stats);
                  }
                  if (cxn != null) {
                      if (procedureInfo != null && procedureInfo.readOnly) {
                          stats.addAffinityRead();
                      }
                      else {
                          stats.addAffinityWrite();
                      }
                  }
                  // account these here because we lose the partition ID and procedure info once we
                  // bust out of this scope.
                  else {
                      if (procedureInfo != null && procedureInfo.readOnly) {
                          stats.addRrRead();
                      }
                      else {
                          stats.addRrWrite();
                      }
                  }
              }
              if (cxn == null) {
                  for (int i=0; i < totalConnections; ++i) {
                      cxn = m_connections.get(Math.abs(++m_nextConnection % totalConnections));
                      if (!cxn.hadBackPressure() || ignoreBackpressure) {
                          // serialize and queue the invocation
                          backpressure = false;
                          break;
                      }
                  }
              }
  
              if (backpressure) {
                  cxn = null;
                  for (ClientStatusListenerExt s : m_listeners) {
                      s.backpressure(true);
                  }
              }
          }
  
          /*
           * Do the heavy weight serialization outside the synchronized block.
           * createWork synchronizes on an individual connection which allows for more concurrency
           */
          if (cxn != null) {
              ByteBuffer buf = null;
              try {
                  buf = serializeSPI(invocation);
              } catch (Exception e) {
                  Throwables.propagate(e);
              }
              cxn.createWork(nowNanos, invocation.getHandle(), invocation.getProcName(), buf, cb, ignoreBackpressure, timeoutNanos);
          }
          if (m_topologyChangeAware) {
              createConnectionsUponTopologyChange();
          }
          return !backpressure;
      }