public static void andAckSet(Position currentPosition, Position otherPosition) {
        if (currentPosition == null || otherPosition == null) {
            return;
        }
        AckSetState currentAckSetState = AckSetStateUtil.getAckSetState(currentPosition);
        AckSetState otherAckSetState = AckSetStateUtil.getAckSetState(otherPosition);
        // The ackSet will be an empty long array after the latest single message of a batch was acked.
        currentAckSetState.setAckSet(andAckSet(currentAckSetState.getAckSet(), otherAckSetState.getAckSet()));
    }

private void handleIndividualAck(TxnID txnID, List<MutablePair<Position, Integer>> positions) {
     ...
     if (pendingAckMessageForCurrentTxn.containsKey(position)) {
          PositionAckSetUtil.andAckSet(pendingAckMessageForCurrentTxn.get(position), position);
      }
     ...
}