@Test public void testInputGateRemovalFromNettyShuffleEnvironment() throws Exception { NettyShuffleEnvironment network = createNettyShuffleEnvironment(); try { int numberOfGates = 10; Map<InputGateID, SingleInputGate> createdInputGatesById = createInputGateWithLocalChannels(network, numberOfGates, 1); assertEquals(numberOfGates, createdInputGatesById.size()); for (InputGateID id : createdInputGatesById.keySet()) { assertThat(network.getInputGate(id).isPresent(), is(true)); createdInputGatesById.get(id).close(); assertThat(network.getInputGate(id).isPresent(), is(false)); } } finally { network.close(); } }
@Override public void close() throws IOException { boolean released = false; synchronized (requestLock) { if (!closeFuture.isDone()) { try { LOG.debug("{}: Releasing {}.", owningTaskName, this); if (retriggerLocalRequestTimer != null) { retriggerLocalRequestTimer.cancel(); } for (InputChannel inputChannel : inputChannels.values()) { try { inputChannel.releaseAllResources(); } catch (IOException e) { LOG.warn("{}: Error during release of channel resources: {}.", owningTaskName, e.getMessage(), e); } } if (bufferPool != null) { bufferPool.lazyDestroy(); } } finally { released = true; closeFuture.complete(null); } } } if (released) { synchronized (inputChannelsWithData) { inputChannelsWithData.notifyAll(); } } }
SingleInputGate extends InputGate { @Override public void close() throws IOException { boolean released = false; synchronized (requestLock) { if (!closeFuture.isDone()) { try { LOG.debug("{}: Releasing {}.", owningTaskName, this); if (retriggerLocalRequestTimer != null) { retriggerLocalRequestTimer.cancel(); } for (InputChannel inputChannel : inputChannels.values()) { try { inputChannel.releaseAllResources(); } catch (IOException e) { LOG.warn("{}: Error during release of channel resources: {}.", owningTaskName, e.getMessage(), e); } } if (bufferPool != null) { bufferPool.lazyDestroy(); } } finally { released = true; closeFuture.complete(null); } } } if (released) { synchronized (inputChannelsWithData) { inputChannelsWithData.notifyAll(); } } } }
SingleInputGate extends InputGate { @Override public void close() throws IOException { boolean released = false; synchronized (requestLock) { if (!closeFuture.isDone()) { try { LOG.debug("{}: Releasing {}.", owningTaskName, this); if (retriggerLocalRequestTimer != null) { retriggerLocalRequestTimer.cancel(); } for (InputChannel inputChannel : inputChannels.values()) { try { inputChannel.releaseAllResources(); } catch (IOException e) { LOG.warn("{}: Error during release of channel resources: {}.", owningTaskName, e.getMessage(), e); } } if (bufferPool != null) { bufferPool.lazyDestroy(); } } finally { released = true; closeFuture.complete(null); } } } if (released) { synchronized (inputChannelsWithData) { inputChannelsWithData.notifyAll(); } } } SingleInputGate(
		String owningTaskName,
		IntermediateDataSetID consumedResultId,
		final ResultPartitionType consumedPartitionType,
		int consumedSubpartitionIndex,
		int numberOfInputChannels,
		PartitionProducerStateProvider partitionProducerStateProvider,
		boolean isCreditBased,
		SupplierWithException<BufferPool, IOException> bufferPoolFactory); }
SingleInputGate extends InputGate { @Override public void close() throws IOException { boolean released = false; synchronized (requestLock) { if (!closeFuture.isDone()) { try { LOG.debug("{}: Releasing {}.", owningTaskName, this); if (retriggerLocalRequestTimer != null) { retriggerLocalRequestTimer.cancel(); } for (InputChannel inputChannel : inputChannels.values()) { try { inputChannel.releaseAllResources(); } catch (IOException e) { LOG.warn("{}: Error during release of channel resources: {}.", owningTaskName, e.getMessage(), e); } } if (bufferPool != null) { bufferPool.lazyDestroy(); } } finally { released = true; closeFuture.complete(null); } } } if (released) { synchronized (inputChannelsWithData) { inputChannelsWithData.notifyAll(); } } } SingleInputGate(
		String owningTaskName,
		IntermediateDataSetID consumedResultId,
		final ResultPartitionType consumedPartitionType,
		int consumedSubpartitionIndex,
		int numberOfInputChannels,
		PartitionProducerStateProvider partitionProducerStateProvider,
		boolean isCreditBased,
		SupplierWithException<BufferPool, IOException> bufferPoolFactory); @Override void setup(); @Override int getNumberOfInputChannels(); IntermediateDataSetID getConsumedResultId(); ResultPartitionType getConsumedPartitionType(); BufferPool getBufferPool(); int getNumberOfQueuedBuffers(); CompletableFuture<Void> getCloseFuture(); void setBufferPool(BufferPool bufferPool); @VisibleForTesting void assignExclusiveSegments(); void setInputChannel(IntermediateResultPartitionID partitionId, InputChannel inputChannel); void updateInputChannel(
			ResourceID localLocation,
			NettyShuffleDescriptor shuffleDescriptor); void retriggerPartitionRequest(IntermediateResultPartitionID partitionId); @Override void close(); @Override boolean isFinished(); @Override Optional<BufferOrEvent> getNext(); @Override Optional<BufferOrEvent> pollNext(); @Override void sendTaskEvent(TaskEvent event); Map<IntermediateResultPartitionID, InputChannel> getInputChannels(); }
SingleInputGate extends InputGate { @Override public void close() throws IOException { boolean released = false; synchronized (requestLock) { if (!closeFuture.isDone()) { try { LOG.debug("{}: Releasing {}.", owningTaskName, this); if (retriggerLocalRequestTimer != null) { retriggerLocalRequestTimer.cancel(); } for (InputChannel inputChannel : inputChannels.values()) { try { inputChannel.releaseAllResources(); } catch (IOException e) { LOG.warn("{}: Error during release of channel resources: {}.", owningTaskName, e.getMessage(), e); } } if (bufferPool != null) { bufferPool.lazyDestroy(); } } finally { released = true; closeFuture.complete(null); } } } if (released) { synchronized (inputChannelsWithData) { inputChannelsWithData.notifyAll(); } } } SingleInputGate(
		String owningTaskName,
		IntermediateDataSetID consumedResultId,
		final ResultPartitionType consumedPartitionType,
		int consumedSubpartitionIndex,
		int numberOfInputChannels,
		PartitionProducerStateProvider partitionProducerStateProvider,
		boolean isCreditBased,
		SupplierWithException<BufferPool, IOException> bufferPoolFactory); @Override void setup(); @Override int getNumberOfInputChannels(); IntermediateDataSetID getConsumedResultId(); ResultPartitionType getConsumedPartitionType(); BufferPool getBufferPool(); int getNumberOfQueuedBuffers(); CompletableFuture<Void> getCloseFuture(); void setBufferPool(BufferPool bufferPool); @VisibleForTesting void assignExclusiveSegments(); void setInputChannel(IntermediateResultPartitionID partitionId, InputChannel inputChannel); void updateInputChannel(
			ResourceID localLocation,
			NettyShuffleDescriptor shuffleDescriptor); void retriggerPartitionRequest(IntermediateResultPartitionID partitionId); @Override void close(); @Override boolean isFinished(); @Override Optional<BufferOrEvent> getNext(); @Override Optional<BufferOrEvent> pollNext(); @Override void sendTaskEvent(TaskEvent event); Map<IntermediateResultPartitionID, InputChannel> getInputChannels();  }