@Test public void testChannelErrorWhileRetriggeringRequest() { final SingleInputGate inputGate = createSingleInputGate(1); final LocalInputChannel localChannel = createLocalInputChannel(inputGate, new ResultPartitionManager()); final Timer timer = new Timer(true) { @Override public void schedule(TimerTask task, long delay) { task.run(); try { localChannel.checkError(); fail("Should throw a PartitionNotFoundException."); } catch (PartitionNotFoundException notFound) { assertThat(localChannel.partitionId, Matchers.is(notFound.getPartitionId())); } catch (IOException ex) { fail("Should throw a PartitionNotFoundException."); } } }; try { localChannel.retriggerSubpartitionRequest(timer, 0); } finally { timer.cancel(); } }
void retriggerSubpartitionRequest(Timer timer, final int subpartitionIndex) { synchronized (requestLock) { checkState(subpartitionView == null, "already requested partition"); timer.schedule(new TimerTask() { @Override public void run() { try { requestSubpartition(subpartitionIndex); } catch (Throwable t) { setError(t); } } }, getCurrentBackoff()); } }
LocalInputChannel extends InputChannel implements BufferAvailabilityListener { void retriggerSubpartitionRequest(Timer timer, final int subpartitionIndex) { synchronized (requestLock) { checkState(subpartitionView == null, "already requested partition"); timer.schedule(new TimerTask() { @Override public void run() { try { requestSubpartition(subpartitionIndex); } catch (Throwable t) { setError(t); } } }, getCurrentBackoff()); } } }
LocalInputChannel extends InputChannel implements BufferAvailabilityListener { void retriggerSubpartitionRequest(Timer timer, final int subpartitionIndex) { synchronized (requestLock) { checkState(subpartitionView == null, "already requested partition"); timer.schedule(new TimerTask() { @Override public void run() { try { requestSubpartition(subpartitionIndex); } catch (Throwable t) { setError(t); } } }, getCurrentBackoff()); } } LocalInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ResultPartitionManager partitionManager,
		TaskEventPublisher taskEventPublisher,
		InputChannelMetrics metrics);  LocalInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ResultPartitionManager partitionManager,
		TaskEventPublisher taskEventPublisher,
		int initialBackoff,
		int maxBackoff,
		InputChannelMetrics metrics); }
LocalInputChannel extends InputChannel implements BufferAvailabilityListener { void retriggerSubpartitionRequest(Timer timer, final int subpartitionIndex) { synchronized (requestLock) { checkState(subpartitionView == null, "already requested partition"); timer.schedule(new TimerTask() { @Override public void run() { try { requestSubpartition(subpartitionIndex); } catch (Throwable t) { setError(t); } } }, getCurrentBackoff()); } } LocalInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ResultPartitionManager partitionManager,
		TaskEventPublisher taskEventPublisher,
		InputChannelMetrics metrics);  LocalInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ResultPartitionManager partitionManager,
		TaskEventPublisher taskEventPublisher,
		int initialBackoff,
		int maxBackoff,
		InputChannelMetrics metrics); @Override void notifyDataAvailable(); @Override int unsynchronizedGetNumberOfQueuedBuffers(); @Override String toString(); }
LocalInputChannel extends InputChannel implements BufferAvailabilityListener { void retriggerSubpartitionRequest(Timer timer, final int subpartitionIndex) { synchronized (requestLock) { checkState(subpartitionView == null, "already requested partition"); timer.schedule(new TimerTask() { @Override public void run() { try { requestSubpartition(subpartitionIndex); } catch (Throwable t) { setError(t); } } }, getCurrentBackoff()); } } LocalInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ResultPartitionManager partitionManager,
		TaskEventPublisher taskEventPublisher,
		InputChannelMetrics metrics);  LocalInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ResultPartitionManager partitionManager,
		TaskEventPublisher taskEventPublisher,
		int initialBackoff,
		int maxBackoff,
		InputChannelMetrics metrics); @Override void notifyDataAvailable(); @Override int unsynchronizedGetNumberOfQueuedBuffers(); @Override String toString();  }