@Test public void testOnFailedPartitionRequestDoesNotBlockNetworkThreads() throws Exception { final long testBlockedWaitTimeoutMillis = 30_000L; final PartitionProducerStateChecker partitionProducerStateChecker = (jobId, intermediateDataSetId, resultPartitionId) -> CompletableFuture.completedFuture(ExecutionState.RUNNING); final NettyShuffleEnvironment shuffleEnvironment = new NettyShuffleEnvironmentBuilder().build(); final Task task = new TestTaskBuilder(shuffleEnvironment) .setPartitionProducerStateChecker(partitionProducerStateChecker) .build(); final SingleInputGate inputGate = new SingleInputGateBuilder() .setPartitionProducerStateProvider(task) .build(); TestTaskBuilder.setTaskState(task, ExecutionState.RUNNING); final OneShotLatch ready = new OneShotLatch(); final OneShotLatch blocker = new OneShotLatch(); final AtomicBoolean timedOutOrInterrupted = new AtomicBoolean(false); final ConnectionManager blockingConnectionManager = new TestingConnectionManager() { @Override public PartitionRequestClient createPartitionRequestClient( ConnectionID connectionId) { ready.trigger(); try { blocker.await(testBlockedWaitTimeoutMillis, TimeUnit.MILLISECONDS); } catch (InterruptedException | TimeoutException e) { timedOutOrInterrupted.set(true); } return new TestingPartitionRequestClient(); } }; final RemoteInputChannel remoteInputChannel = InputChannelBuilder.newBuilder() .setConnectionManager(blockingConnectionManager) .buildRemoteAndSetToGate(inputGate); final Thread simulatedNetworkThread = new Thread( () -> { try { ready.await(); remoteInputChannel.onFailedPartitionRequest(); blocker.trigger(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); simulatedNetworkThread.start(); inputGate.requestPartitions(); simulatedNetworkThread.join(); Assert.assertFalse( "Test ended by timeout or interruption - this indicates that the network thread was blocked.", timedOutOrInterrupted.get()); }
public void onFailedPartitionRequest() { inputGate.triggerPartitionStateCheck(partitionId); }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onFailedPartitionRequest() { inputGate.triggerPartitionStateCheck(partitionId); } }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onFailedPartitionRequest() { inputGate.triggerPartitionStateCheck(partitionId); } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		InputChannelMetrics metrics,
		@Nonnull MemorySegmentProvider memorySegmentProvider); }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onFailedPartitionRequest() { inputGate.triggerPartitionStateCheck(partitionId); } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		InputChannelMetrics metrics,
		@Nonnull MemorySegmentProvider memorySegmentProvider); @VisibleForTesting @Override void requestSubpartition(int subpartitionIndex); @Override boolean isReleased(); @Override String toString(); @Override void recycle(MemorySegment segment); int getNumberOfAvailableBuffers(); int getNumberOfRequiredBuffers(); int getSenderBacklog(); @Override NotificationResult notifyBufferAvailable(Buffer buffer); @Override void notifyBufferDestroyed(); int getUnannouncedCredit(); int getAndResetUnannouncedCredit(); int getNumberOfQueuedBuffers(); @Override int unsynchronizedGetNumberOfQueuedBuffers(); int unsynchronizedGetExclusiveBuffersUsed(); int unsynchronizedGetFloatingBuffersAvailable(); InputChannelID getInputChannelId(); int getInitialCredit(); BufferProvider getBufferProvider(); @Nullable Buffer requestBuffer(); void onBuffer(Buffer buffer, int sequenceNumber, int backlog); void onEmptyBuffer(int sequenceNumber, int backlog); void onFailedPartitionRequest(); void onError(Throwable cause); }
RemoteInputChannel extends InputChannel implements BufferRecycler, BufferListener { public void onFailedPartitionRequest() { inputGate.triggerPartitionStateCheck(partitionId); } RemoteInputChannel(
		SingleInputGate inputGate,
		int channelIndex,
		ResultPartitionID partitionId,
		ConnectionID connectionId,
		ConnectionManager connectionManager,
		int initialBackOff,
		int maxBackoff,
		InputChannelMetrics metrics,
		@Nonnull MemorySegmentProvider memorySegmentProvider); @VisibleForTesting @Override void requestSubpartition(int subpartitionIndex); @Override boolean isReleased(); @Override String toString(); @Override void recycle(MemorySegment segment); int getNumberOfAvailableBuffers(); int getNumberOfRequiredBuffers(); int getSenderBacklog(); @Override NotificationResult notifyBufferAvailable(Buffer buffer); @Override void notifyBufferDestroyed(); int getUnannouncedCredit(); int getAndResetUnannouncedCredit(); int getNumberOfQueuedBuffers(); @Override int unsynchronizedGetNumberOfQueuedBuffers(); int unsynchronizedGetExclusiveBuffersUsed(); int unsynchronizedGetFloatingBuffersAvailable(); InputChannelID getInputChannelId(); int getInitialCredit(); BufferProvider getBufferProvider(); @Nullable Buffer requestBuffer(); void onBuffer(Buffer buffer, int sequenceNumber, int backlog); void onEmptyBuffer(int sequenceNumber, int backlog); void onFailedPartitionRequest(); void onError(Throwable cause);  }