docs: private Future<Message> add(StaticBuffer content, int partitionId, ExternalPersistor persistor) {
          ResourceUnavailableException.verifyOpen(isOpen,"Log",name);
          Preconditions.checkArgument(content!=null && content.length()>0,"Content is empty");
          Preconditions.checkArgument(partitionId>=0 && partitionId<(1<<manager.partitionBitWidth),"Invalid partition id: %s",partitionId);
          final Instant timestamp = times.getTime();
          KCVSMessage msg = new KCVSMessage(content,timestamp,manager.senderId);
          FutureMessage fmsg = new FutureMessage(msg);
  
          StaticBuffer key=getLogKey(partitionId,(int)(numBucketCounter.incrementAndGet()%numBuckets),getTimeSlice(timestamp));
          MessageEnvelope envelope = new MessageEnvelope(fmsg,key,writeMessage(msg));
  
          if (persistor!=null) {
              try {
                  persistor.add(envelope.key,envelope.entry);
                  envelope.message.delivered();
              } catch (TitanException e) {
                  envelope.message.failed(e);
                  throw e;
              }
          } else if (outgoingMsg==null) {
              sendMessages(ImmutableList.of(envelope));
          } else {
              try {
                  outgoingMsg.put(envelope); //Produces back pressure when full
                  log.debug("Enqueued {} for partition {}", envelope, partitionId);
              } catch (InterruptedException e) {
                  throw new TitanException("Got interrupted waiting to send message",e);
              }
          }
          return fmsg;
      }