test_class.file: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueueTest.java
test_class.identifier: PartitionRequestQueueTest
test_case.body: @Test
  	public void testProducerFailedException() throws Exception {
  		PartitionRequestQueue queue = new PartitionRequestQueue();
  
  		EmbeddedChannel ch = new EmbeddedChannel(queue);
  
  		ResultSubpartitionView view = mock(ResultSubpartitionView.class);
  		when(view.isReleased()).thenReturn(true);
  		when(view.getFailureCause()).thenReturn(new RuntimeException("Expected test exception"));
  
  		// Enqueue the erroneous view
  		queue.enqueue(view, new InputChannelID());
  		ch.runPendingTasks();
  
  		// Read the enqueued msg
  		Object msg = ch.readOutbound();
  
  		assertEquals(msg.getClass(), NettyMessage.ErrorResponse.class);
  
  		NettyMessage.ErrorResponse err = (NettyMessage.ErrorResponse) msg;
  		assertTrue(err.cause instanceof CancelTaskException);
  	}
test_case.class_method_signature: PartitionRequestQueueTest.testProducerFailedException()
test_case.full_signature: @Test public void testProducerFailedException()
test_case.identifier: testProducerFailedException
test_case.invocations: mock thenReturn when isReleased thenReturn when getFailureCause enqueue runPendingTasks readOutbound assertEquals getClass assertTrue
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testProducerFailedException()
focal_class.declarator: LOG = LoggerFactory.getLogger(PartitionRequestQueue.class)
focal_class.modifier: private final
focal_class.original_string: private final Logger LOG = LoggerFactory.getLogger(PartitionRequestQueue.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: writeListener = new WriteAndFlushNextMessageIfPossibleListener()
focal_class.modifier: private final
focal_class.original_string: private final ChannelFutureListener writeListener = new WriteAndFlushNextMessageIfPossibleListener();
focal_class.type: ChannelFutureListener
focal_class.var_name: writeListener
focal_class.declarator: queue = new ArrayDeque<SequenceNumberingSubpartitionView>()
focal_class.modifier: private final
focal_class.original_string: private final Queue<SequenceNumberingSubpartitionView> queue = new ArrayDeque<SequenceNumberingSubpartitionView>();
focal_class.type: Queue<SequenceNumberingSubpartitionView>
focal_class.var_name: queue
focal_class.declarator: released = Sets.newHashSet()
focal_class.modifier: private final
focal_class.original_string: private final Set<InputChannelID> released = Sets.newHashSet();
focal_class.type: Set<InputChannelID>
focal_class.var_name: released
focal_class.declarator: currentPartitionQueue
focal_class.modifier: private
focal_class.original_string: private SequenceNumberingSubpartitionView currentPartitionQueue;
focal_class.type: SequenceNumberingSubpartitionView
focal_class.var_name: currentPartitionQueue
focal_class.declarator: fatalError
focal_class.modifier: private
focal_class.original_string: private boolean fatalError;
focal_class.type: boolean
focal_class.var_name: fatalError
focal_class.declarator: ctx
focal_class.modifier: private
focal_class.original_string: private ChannelHandlerContext ctx;
focal_class.type: ChannelHandlerContext
focal_class.var_name: ctx
focal_class.file: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/netty/PartitionRequestQueue.java
focal_class.identifier: PartitionRequestQueue
focal_class.class_method_signature: PartitionRequestQueue.channelRegistered(final ChannelHandlerContext ctx)
focal_class.full_signature: @Override public void channelRegistered(final ChannelHandlerContext ctx)
focal_class.identifier: channelRegistered
focal_class.modifiers: @Override public
focal_class.parameters: (final ChannelHandlerContext ctx)
focal_class.return: void
focal_class.signature: void channelRegistered(final ChannelHandlerContext ctx)
focal_class.class_method_signature: PartitionRequestQueue.enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId)
focal_class.full_signature: public void enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId)
focal_class.identifier: enqueue
focal_class.modifiers: public
focal_class.parameters: (ResultSubpartitionView partitionQueue, InputChannelID receiverId)
focal_class.return: void
focal_class.signature: void enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId)
focal_class.class_method_signature: PartitionRequestQueue.cancel(InputChannelID receiverId)
focal_class.full_signature: public void cancel(InputChannelID receiverId)
focal_class.identifier: cancel
focal_class.modifiers: public
focal_class.parameters: (InputChannelID receiverId)
focal_class.return: void
focal_class.signature: void cancel(InputChannelID receiverId)
focal_class.class_method_signature: PartitionRequestQueue.close()
focal_class.full_signature: public void close()
focal_class.identifier: close
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: PartitionRequestQueue.userEventTriggered(ChannelHandlerContext ctx, Object msg)
focal_class.full_signature: @Override public void userEventTriggered(ChannelHandlerContext ctx, Object msg)
focal_class.identifier: userEventTriggered
focal_class.modifiers: @Override public
focal_class.parameters: (ChannelHandlerContext ctx, Object msg)
focal_class.return: void
focal_class.signature: void userEventTriggered(ChannelHandlerContext ctx, Object msg)
focal_class.class_method_signature: PartitionRequestQueue.channelWritabilityChanged(ChannelHandlerContext ctx)
focal_class.full_signature: @Override public void channelWritabilityChanged(ChannelHandlerContext ctx)
focal_class.identifier: channelWritabilityChanged
focal_class.modifiers: @Override public
focal_class.parameters: (ChannelHandlerContext ctx)
focal_class.return: void
focal_class.signature: void channelWritabilityChanged(ChannelHandlerContext ctx)
focal_class.class_method_signature: PartitionRequestQueue.writeAndFlushNextMessageIfPossible(final Channel channel)
focal_class.full_signature: private void writeAndFlushNextMessageIfPossible(final Channel channel)
focal_class.identifier: writeAndFlushNextMessageIfPossible
focal_class.modifiers: private
focal_class.parameters: (final Channel channel)
focal_class.return: void
focal_class.signature: void writeAndFlushNextMessageIfPossible(final Channel channel)
focal_class.class_method_signature: PartitionRequestQueue.channelInactive(ChannelHandlerContext ctx)
focal_class.full_signature: @Override public void channelInactive(ChannelHandlerContext ctx)
focal_class.identifier: channelInactive
focal_class.modifiers: @Override public
focal_class.parameters: (ChannelHandlerContext ctx)
focal_class.return: void
focal_class.signature: void channelInactive(ChannelHandlerContext ctx)
focal_class.class_method_signature: PartitionRequestQueue.exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
focal_class.full_signature: @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
focal_class.identifier: exceptionCaught
focal_class.modifiers: @Override public
focal_class.parameters: (ChannelHandlerContext ctx, Throwable cause)
focal_class.return: void
focal_class.signature: void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
focal_class.class_method_signature: PartitionRequestQueue.handleException(Channel channel, Throwable cause)
focal_class.full_signature: private void handleException(Channel channel, Throwable cause)
focal_class.identifier: handleException
focal_class.modifiers: private
focal_class.parameters: (Channel channel, Throwable cause)
focal_class.return: void
focal_class.signature: void handleException(Channel channel, Throwable cause)
focal_class.class_method_signature: PartitionRequestQueue.releaseAllResources()
focal_class.full_signature: private void releaseAllResources()
focal_class.identifier: releaseAllResources
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void releaseAllResources()
focal_class.class_method_signature: PartitionRequestQueue.markAsReleased(InputChannelID receiverId)
focal_class.full_signature: private void markAsReleased(InputChannelID receiverId)
focal_class.identifier: markAsReleased
focal_class.modifiers: private
focal_class.parameters: (InputChannelID receiverId)
focal_class.return: void
focal_class.signature: void markAsReleased(InputChannelID receiverId)
focal_class.superclass: extends ChannelInboundHandlerAdapter
focal_method.body: public void enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId) throws Exception {
  		ctx.pipeline().fireUserEventTriggered(new SequenceNumberingSubpartitionView(partitionQueue, receiverId));
  	}
focal_method.class_method_signature: PartitionRequestQueue.enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId)
focal_method.full_signature: public void enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId)
focal_method.identifier: enqueue
focal_method.invocations: fireUserEventTriggered pipeline
focal_method.modifiers: public
focal_method.parameters: (ResultSubpartitionView partitionQueue, InputChannelID receiverId)
focal_method.return: void
focal_method.signature: void enqueue(ResultSubpartitionView partitionQueue, InputChannelID receiverId)