code: private <T> ListenableFuture<T> issueCommandToCluster(final KayVeeCommand kayVeeCommand) {
          final SettableFuture<T> returned = SettableFuture.create();
  
          try {
              PendingCommandData previous = pendingCommands.put(kayVeeCommand.getCommandId(), new PendingCommandData(returned));
              checkState(previous == null, "existing command:%s", previous);
  
              Futures.addCallback(returned, new FutureCallback<Object>() {
                  @Override
                  public void onSuccess(Object result) {
                      // success handled in the apply command block
                  }
  
                  @Override
                  public void onFailure(Throwable t) {
                      pendingCommands.remove(kayVeeCommand.getCommandId());
                  }
              });
  
              // it's possible for this to throw an IllegalStateException
              // if another thread calls "stop" while a request is just about to be submitted
              // since access to raftAgent itself is not synchronized here
              ListenableFuture<Void> consensusFuture = raftAgent.submitCommand(kayVeeCommand);
              Futures.addCallback(consensusFuture, new FutureCallback<Void>() {
                  @Override
                  public void onSuccess(Void result) {
                      // noop - we wait until we attempt to apply the command locally
                  }
  
                  @Override
                  public void onFailure(Throwable t) {
                      returned.setException(t);
                  }
              });
          } catch (NotLeaderException e) {
              returned.setException(e);
          }
  
          return returned;
      }
docs: IMPORTANT: DO NOT HOLD A LOCK WHEN CALLING issueCommandToCluster OR IN THE onLeadershipChange CALLBACK