[TEZ-1036] Shuffle deduping should be global and not local to the MapHost

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Labels: newbie++
• Assignee: Unassigned
• Reporter: Bikas Saha
• Created: 10/Apr/14 18:13
• Updated: 10/Apr/14 23:17

Description

The following logic needs to be across all fetchers and not within a maphost.

List<InputAttemptIdentifier> origList = host.getAndClearKnownMaps();

    Map<Integer, InputAttemptIdentifier> dedupedList = new LinkedHashMap<Integer, InputAttemptIdentifier>();
    Iterator<InputAttemptIdentifier> listItr = origList.iterator();
    while (listItr.hasNext()) {
      // we may want to try all versions of the input but with current retry
      // behavior older ones are likely to be lost and should be ignored.
      // This may be removed after TEZ-914
      InputAttemptIdentifier id = listItr.next();
      if (inputShouldBeConsumed(id)) {
        Integer inputNumber = new Integer(id.getInputIdentifier().getInputIndex());
        InputAttemptIdentifier oldId = dedupedList.get(inputNumber);
        if (oldId == null || oldId.getAttemptNumber() < id.getAttemptNumber()) {
          dedupedList.put(inputNumber, id);
          if (oldId != null) {
            LOG.warn("Old Src for InputIndex: " + inputNumber + " with attemptNumber: "
                + oldId.getAttemptNumber()
                + " was not determined to be invalid. Ignoring it for now in favour of "
                + id.getAttemptNumber());
          }
        }
      } else {
        LOG.info("Ignoring finished or obsolete source: " + id);
      }
    }