@Test public void testUpdateChannelBeforeRequest() throws Exception { SingleInputGate inputGate = createInputGate(1); ResultPartitionManager partitionManager = mock(ResultPartitionManager.class); InputChannel unknown = InputChannelBuilder.newBuilder() .setPartitionManager(partitionManager) .buildUnknownAndSetToGate(inputGate); ResultPartitionID resultPartitionID = unknown.getPartitionId(); ResourceID location = ResourceID.generate(); inputGate.updateInputChannel(location, createRemoteWithIdAndLocation(resultPartitionID.getPartitionId(), location)); verify(partitionManager, never()).createSubpartitionView( any(ResultPartitionID.class), anyInt(), any(BufferAvailabilityListener.class)); }
public void updateInputChannel( ResourceID localLocation, NettyShuffleDescriptor shuffleDescriptor) throws IOException, InterruptedException { synchronized (requestLock) { if (closeFuture.isDone()) { return; } IntermediateResultPartitionID partitionId = shuffleDescriptor.getResultPartitionID().getPartitionId(); InputChannel current = inputChannels.get(partitionId); if (current instanceof UnknownInputChannel) { UnknownInputChannel unknownChannel = (UnknownInputChannel) current; boolean isLocal = shuffleDescriptor.isLocalTo(localLocation); InputChannel newChannel; if (isLocal) { newChannel = unknownChannel.toLocalInputChannel(); } else { RemoteInputChannel remoteInputChannel = unknownChannel.toRemoteInputChannel(shuffleDescriptor.getConnectionId()); if (isCreditBased) { remoteInputChannel.assignExclusiveSegments(); } newChannel = remoteInputChannel; } LOG.debug("{}: Updated unknown input channel to {}.", owningTaskName, newChannel); inputChannels.put(partitionId, newChannel); if (requestedPartitionsFlag) { newChannel.requestSubpartition(consumedSubpartitionIndex); } for (TaskEvent event : pendingEvents) { newChannel.sendTaskEvent(event); } if (--numberOfUninitializedChannels == 0) { pendingEvents.clear(); } } } }
SingleInputGate extends InputGate { public void updateInputChannel( ResourceID localLocation, NettyShuffleDescriptor shuffleDescriptor) throws IOException, InterruptedException { synchronized (requestLock) { if (closeFuture.isDone()) { return; } IntermediateResultPartitionID partitionId = shuffleDescriptor.getResultPartitionID().getPartitionId(); InputChannel current = inputChannels.get(partitionId); if (current instanceof UnknownInputChannel) { UnknownInputChannel unknownChannel = (UnknownInputChannel) current; boolean isLocal = shuffleDescriptor.isLocalTo(localLocation); InputChannel newChannel; if (isLocal) { newChannel = unknownChannel.toLocalInputChannel(); } else { RemoteInputChannel remoteInputChannel = unknownChannel.toRemoteInputChannel(shuffleDescriptor.getConnectionId()); if (isCreditBased) { remoteInputChannel.assignExclusiveSegments(); } newChannel = remoteInputChannel; } LOG.debug("{}: Updated unknown input channel to {}.", owningTaskName, newChannel); inputChannels.put(partitionId, newChannel); if (requestedPartitionsFlag) { newChannel.requestSubpartition(consumedSubpartitionIndex); } for (TaskEvent event : pendingEvents) { newChannel.sendTaskEvent(event); } if (--numberOfUninitializedChannels == 0) { pendingEvents.clear(); } } } } }
SingleInputGate extends InputGate { public void updateInputChannel( ResourceID localLocation, NettyShuffleDescriptor shuffleDescriptor) throws IOException, InterruptedException { synchronized (requestLock) { if (closeFuture.isDone()) { return; } IntermediateResultPartitionID partitionId = shuffleDescriptor.getResultPartitionID().getPartitionId(); InputChannel current = inputChannels.get(partitionId); if (current instanceof UnknownInputChannel) { UnknownInputChannel unknownChannel = (UnknownInputChannel) current; boolean isLocal = shuffleDescriptor.isLocalTo(localLocation); InputChannel newChannel; if (isLocal) { newChannel = unknownChannel.toLocalInputChannel(); } else { RemoteInputChannel remoteInputChannel = unknownChannel.toRemoteInputChannel(shuffleDescriptor.getConnectionId()); if (isCreditBased) { remoteInputChannel.assignExclusiveSegments(); } newChannel = remoteInputChannel; } LOG.debug("{}: Updated unknown input channel to {}.", owningTaskName, newChannel); inputChannels.put(partitionId, newChannel); if (requestedPartitionsFlag) { newChannel.requestSubpartition(consumedSubpartitionIndex); } for (TaskEvent event : pendingEvents) { newChannel.sendTaskEvent(event); } if (--numberOfUninitializedChannels == 0) { pendingEvents.clear(); } } } } SingleInputGate(
		String owningTaskName,
		IntermediateDataSetID consumedResultId,
		final ResultPartitionType consumedPartitionType,
		int consumedSubpartitionIndex,
		int numberOfInputChannels,
		PartitionProducerStateProvider partitionProducerStateProvider,
		boolean isCreditBased,
		SupplierWithException<BufferPool, IOException> bufferPoolFactory); }
SingleInputGate extends InputGate { public void updateInputChannel( ResourceID localLocation, NettyShuffleDescriptor shuffleDescriptor) throws IOException, InterruptedException { synchronized (requestLock) { if (closeFuture.isDone()) { return; } IntermediateResultPartitionID partitionId = shuffleDescriptor.getResultPartitionID().getPartitionId(); InputChannel current = inputChannels.get(partitionId); if (current instanceof UnknownInputChannel) { UnknownInputChannel unknownChannel = (UnknownInputChannel) current; boolean isLocal = shuffleDescriptor.isLocalTo(localLocation); InputChannel newChannel; if (isLocal) { newChannel = unknownChannel.toLocalInputChannel(); } else { RemoteInputChannel remoteInputChannel = unknownChannel.toRemoteInputChannel(shuffleDescriptor.getConnectionId()); if (isCreditBased) { remoteInputChannel.assignExclusiveSegments(); } newChannel = remoteInputChannel; } LOG.debug("{}: Updated unknown input channel to {}.", owningTaskName, newChannel); inputChannels.put(partitionId, newChannel); if (requestedPartitionsFlag) { newChannel.requestSubpartition(consumedSubpartitionIndex); } for (TaskEvent event : pendingEvents) { newChannel.sendTaskEvent(event); } if (--numberOfUninitializedChannels == 0) { pendingEvents.clear(); } } } } 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 { public void updateInputChannel( ResourceID localLocation, NettyShuffleDescriptor shuffleDescriptor) throws IOException, InterruptedException { synchronized (requestLock) { if (closeFuture.isDone()) { return; } IntermediateResultPartitionID partitionId = shuffleDescriptor.getResultPartitionID().getPartitionId(); InputChannel current = inputChannels.get(partitionId); if (current instanceof UnknownInputChannel) { UnknownInputChannel unknownChannel = (UnknownInputChannel) current; boolean isLocal = shuffleDescriptor.isLocalTo(localLocation); InputChannel newChannel; if (isLocal) { newChannel = unknownChannel.toLocalInputChannel(); } else { RemoteInputChannel remoteInputChannel = unknownChannel.toRemoteInputChannel(shuffleDescriptor.getConnectionId()); if (isCreditBased) { remoteInputChannel.assignExclusiveSegments(); } newChannel = remoteInputChannel; } LOG.debug("{}: Updated unknown input channel to {}.", owningTaskName, newChannel); inputChannels.put(partitionId, newChannel); if (requestedPartitionsFlag) { newChannel.requestSubpartition(consumedSubpartitionIndex); } for (TaskEvent event : pendingEvents) { newChannel.sendTaskEvent(event); } if (--numberOfUninitializedChannels == 0) { pendingEvents.clear(); } } } } 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();  }