### Question:
LocalInputChannel extends InputChannel implements BufferAvailabilityListener { @Override void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException { boolean retriggerRequest = false; synchronized (requestLock) { checkState(!isReleased, "LocalInputChannel has been released already"); if (subpartitionView == null) { LOG.debug("{}: Requesting LOCAL subpartition {} of partition {}.", this, subpartitionIndex, partitionId); try { ResultSubpartitionView subpartitionView = partitionManager.createSubpartitionView( partitionId, subpartitionIndex, this); if (subpartitionView == null) { throw new IOException("Error requesting subpartition."); } this.subpartitionView = subpartitionView; if (isReleased) { subpartitionView.releaseAllResources(); this.subpartitionView = null; } } catch (PartitionNotFoundException notFound) { if (increaseBackoff()) { retriggerRequest = true; } else { throw notFound; } } } } if (retriggerRequest) { inputGate.retriggerPartitionRequest(partitionId.getPartitionId()); } } 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();  }### Answer:
@Test public void testPartitionNotFoundExceptionWhileRequestingPartition() throws Exception { final SingleInputGate inputGate = createSingleInputGate(1); final LocalInputChannel localChannel = createLocalInputChannel(inputGate, new ResultPartitionManager()); try { localChannel.requestSubpartition(0); fail("Should throw a PartitionNotFoundException."); } catch (PartitionNotFoundException notFound) { assertThat(localChannel.getPartitionId(), Matchers.is(notFound.getPartitionId())); } }
@Test public void testRetriggerPartitionRequestWhilePartitionNotFound() throws Exception { final SingleInputGate inputGate = createSingleInputGate(1); final LocalInputChannel localChannel = createLocalInputChannel( inputGate, new ResultPartitionManager(), 1, 1); inputGate.setInputChannel(localChannel.getPartitionId().getPartitionId(), localChannel); localChannel.requestSubpartition(0); assertNotNull(inputGate.getRetriggerLocalRequestTimer()); }
@Test public void testConcurrentReleaseAndRetriggerPartitionRequest() throws Exception { final SingleInputGate gate = createSingleInputGate(1); ResultPartitionManager partitionManager = mock(ResultPartitionManager.class); when(partitionManager .createSubpartitionView( any(ResultPartitionID.class), anyInt(), any(BufferAvailabilityListener.class))) .thenAnswer(new Answer<ResultSubpartitionView>() { @Override public ResultSubpartitionView answer(InvocationOnMock invocationOnMock) throws Throwable { Thread.sleep(100); throw new PartitionNotFoundException(new ResultPartitionID()); } }); final LocalInputChannel channel = createLocalInputChannel(gate, partitionManager, 1, 1); Thread releaser = new Thread() { @Override public void run() { try { gate.close(); } catch (IOException ignored) { } } }; Thread requester = new Thread() { @Override public void run() { try { channel.requestSubpartition(0); } catch (IOException | InterruptedException ignored) { } } }; requester.start(); releaser.start(); releaser.join(); requester.join(); }