@Test public void testFetchedEventsModifyUnderlyingList() { List<TezEvent> fetchedList; TezTaskID taskId = getNewTaskID(); scheduleTaskAttempt(taskId); sendTezEventsToTask(taskId, 2); TezTaskAttemptID attemptID = mockTask.getAttemptList().iterator().next() .getID(); fetchedList = mockTask.getTaskAttemptTezEvents(attemptID, 0, 100); assertEquals(2, fetchedList.size()); sendTezEventsToTask(taskId, 4); assertEquals(2, fetchedList.size()); fetchedList = mockTask.getTaskAttemptTezEvents(attemptID, 0, 100); assertEquals(6, fetchedList.size()); }
@Override public List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, int fromEventId, int maxEvents) { List<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; readLock.lock(); if (!attempts.containsKey(attemptID)) { throw new TezUncheckedException("Unknown TA: " + attemptID + " asking for events from task:" + getTaskId()); } try { if (tezEventsForTaskAttempts.size() > fromEventId) { int actualMax = Math.min(maxEvents, (tezEventsForTaskAttempts.size() - fromEventId)); int toEventId = actualMax + fromEventId; events = Collections.unmodifiableList(new ArrayList<TezEvent>( tezEventsForTaskAttempts.subList(fromEventId, toEventId))); LOG.info("TaskAttempt:" + attemptID + " sent events: (" + fromEventId + "-" + toEventId + ")"); } return events; } finally { readLock.unlock(); } }
TaskImpl implements Task, EventHandler<TaskEvent> { @Override public List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, int fromEventId, int maxEvents) { List<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; readLock.lock(); if (!attempts.containsKey(attemptID)) { throw new TezUncheckedException("Unknown TA: " + attemptID + " asking for events from task:" + getTaskId()); } try { if (tezEventsForTaskAttempts.size() > fromEventId) { int actualMax = Math.min(maxEvents, (tezEventsForTaskAttempts.size() - fromEventId)); int toEventId = actualMax + fromEventId; events = Collections.unmodifiableList(new ArrayList<TezEvent>( tezEventsForTaskAttempts.subList(fromEventId, toEventId))); LOG.info("TaskAttempt:" + attemptID + " sent events: (" + fromEventId + "-" + toEventId + ")"); } return events; } finally { readLock.unlock(); } } }
TaskImpl implements Task, EventHandler<TaskEvent> { @Override public List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, int fromEventId, int maxEvents) { List<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; readLock.lock(); if (!attempts.containsKey(attemptID)) { throw new TezUncheckedException("Unknown TA: " + attemptID + " asking for events from task:" + getTaskId()); } try { if (tezEventsForTaskAttempts.size() > fromEventId) { int actualMax = Math.min(maxEvents, (tezEventsForTaskAttempts.size() - fromEventId)); int toEventId = actualMax + fromEventId; events = Collections.unmodifiableList(new ArrayList<TezEvent>( tezEventsForTaskAttempts.subList(fromEventId, toEventId))); LOG.info("TaskAttempt:" + attemptID + " sent events: (" + fromEventId + "-" + toEventId + ")"); } return events; } finally { readLock.unlock(); } } TaskImpl(TezVertexID vertexId, int taskIndex,
      EventHandler eventHandler, Configuration conf,
      TaskAttemptListener taskAttemptListener,
      Clock clock, TaskHeartbeatHandler thh, AppContext appContext,
      boolean leafVertex, Resource resource,
      ContainerContext containerContext); }
TaskImpl implements Task, EventHandler<TaskEvent> { @Override public List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, int fromEventId, int maxEvents) { List<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; readLock.lock(); if (!attempts.containsKey(attemptID)) { throw new TezUncheckedException("Unknown TA: " + attemptID + " asking for events from task:" + getTaskId()); } try { if (tezEventsForTaskAttempts.size() > fromEventId) { int actualMax = Math.min(maxEvents, (tezEventsForTaskAttempts.size() - fromEventId)); int toEventId = actualMax + fromEventId; events = Collections.unmodifiableList(new ArrayList<TezEvent>( tezEventsForTaskAttempts.subList(fromEventId, toEventId))); LOG.info("TaskAttempt:" + attemptID + " sent events: (" + fromEventId + "-" + toEventId + ")"); } return events; } finally { readLock.unlock(); } } TaskImpl(TezVertexID vertexId, int taskIndex,
      EventHandler eventHandler, Configuration conf,
      TaskAttemptListener taskAttemptListener,
      Clock clock, TaskHeartbeatHandler thh, AppContext appContext,
      boolean leafVertex, Resource resource,
      ContainerContext containerContext); @Override TaskState getState(); @Override Map<TezTaskAttemptID, TaskAttempt> getAttempts(); @Override TaskAttempt getAttempt(TezTaskAttemptID attemptID); @Override Vertex getVertex(); @Override TezTaskID getTaskId(); @Override boolean isFinished(); @Override TaskReport getReport(); @Override TezCounters getCounters(); @Override float getProgress(); @Override List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID,
      int fromEventId, int maxEvents); @Override List<String> getDiagnostics(); @Override TaskState restoreFromEvent(HistoryEvent historyEvent); @VisibleForTesting TaskStateInternal getInternalState(); @Override boolean canCommit(TezTaskAttemptID taskAttemptID); @Override boolean needsWaitAfterOutputConsumable(); @Override TezTaskAttemptID getOutputConsumableAttempt(); @Override TaskAttempt getSuccessfulAttempt(); @Override void handle(TaskEvent event); }
TaskImpl implements Task, EventHandler<TaskEvent> { @Override public List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, int fromEventId, int maxEvents) { List<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; readLock.lock(); if (!attempts.containsKey(attemptID)) { throw new TezUncheckedException("Unknown TA: " + attemptID + " asking for events from task:" + getTaskId()); } try { if (tezEventsForTaskAttempts.size() > fromEventId) { int actualMax = Math.min(maxEvents, (tezEventsForTaskAttempts.size() - fromEventId)); int toEventId = actualMax + fromEventId; events = Collections.unmodifiableList(new ArrayList<TezEvent>( tezEventsForTaskAttempts.subList(fromEventId, toEventId))); LOG.info("TaskAttempt:" + attemptID + " sent events: (" + fromEventId + "-" + toEventId + ")"); } return events; } finally { readLock.unlock(); } } TaskImpl(TezVertexID vertexId, int taskIndex,
      EventHandler eventHandler, Configuration conf,
      TaskAttemptListener taskAttemptListener,
      Clock clock, TaskHeartbeatHandler thh, AppContext appContext,
      boolean leafVertex, Resource resource,
      ContainerContext containerContext); @Override TaskState getState(); @Override Map<TezTaskAttemptID, TaskAttempt> getAttempts(); @Override TaskAttempt getAttempt(TezTaskAttemptID attemptID); @Override Vertex getVertex(); @Override TezTaskID getTaskId(); @Override boolean isFinished(); @Override TaskReport getReport(); @Override TezCounters getCounters(); @Override float getProgress(); @Override List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID,
      int fromEventId, int maxEvents); @Override List<String> getDiagnostics(); @Override TaskState restoreFromEvent(HistoryEvent historyEvent); @VisibleForTesting TaskStateInternal getInternalState(); @Override boolean canCommit(TezTaskAttemptID taskAttemptID); @Override boolean needsWaitAfterOutputConsumable(); @Override TezTaskAttemptID getOutputConsumableAttempt(); @Override TaskAttempt getSuccessfulAttempt(); @Override void handle(TaskEvent event);  }