docs: protected void processAcksAndFails(int srcTaskId, int taskId,
                                       HeronTuples.HeronControlTupleSet controlTupleSet) {
      HeronTuples.HeronTupleSet.Builder out = HeronTuples.HeronTupleSet.newBuilder();
      out.setSrcTaskId(srcTaskId);
  
      // First go over emits. This makes sure that new emits makes
      // a tuples stay alive before we process its acks
      for (HeronTuples.AckTuple emitTuple : controlTupleSet.getEmitsList()) {
        for (HeronTuples.RootId rootId : emitTuple.getRootsList()) {
          xorManager.anchor(taskId, rootId.getKey(), emitTuple.getAckedtuple());
        }
      }
  
      // Then go over acks
      for (HeronTuples.AckTuple ackTuple : controlTupleSet.getAcksList()) {
        for (HeronTuples.RootId rootId : ackTuple.getRootsList()) {
          if (xorManager.anchor(taskId, rootId.getKey(), ackTuple.getAckedtuple())) {
            // This tuple tree is all over
  
            HeronTuples.AckTuple.Builder a = out.getControlBuilder().addAcksBuilder();
            HeronTuples.RootId.Builder r = a.addRootsBuilder();
  
            r.setKey(rootId.getKey());
            r.setTaskid(taskId);
  
            a.setAckedtuple(0); //  This is ignored
  
            xorManager.remove(taskId, rootId.getKey());
          }
        }
      }
  
      // Now go over the fails
      for (HeronTuples.AckTuple failTuple : controlTupleSet.getFailsList()) {
        for (HeronTuples.RootId rootId : failTuple.getRootsList()) {
          if (xorManager.remove(taskId, rootId.getKey())) {
            // This tuple tree is failed
  
            HeronTuples.AckTuple.Builder f =
                out.getControlBuilder().addFailsBuilder();
            HeronTuples.RootId.Builder r = f.addRootsBuilder();
  
            r.setKey(rootId.getKey());
            r.setTaskid(taskId);
            f.setAckedtuple(0); //  This is ignored
          }
        }
      }
  
      // Check if we need to send ack tuples to spout task
      if (out.hasControl()) {
        sendMessageToInstance(taskId, out.build());
      }
    }