Github user NicoK commented on a diff in the pull request: https://github.com/apache/flink/pull/4533#discussion_r154322427 --- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientHandlerTest.java --- @@ -244,54 +237,89 @@ public void testCancelBeforeActive() throws Exception { } /** - * Verifies that {@link RemoteInputChannel} is enqueued in the pipeline, and - * {@link AddCredit} message is sent to the producer. + * Verifies that {@link RemoteInputChannel} is enqueued in the pipeline for notifying credits, + * and verifies the behaviour of credit notification by triggering channel's writability changed. */ @Test public void testNotifyCreditAvailable() throws Exception { + final SingleInputGate inputGate = createSingleInputGate(); + final RemoteInputChannel inputChannel1 = spy(createRemoteInputChannel(inputGate)); + final RemoteInputChannel inputChannel2 = spy(createRemoteInputChannel(inputGate)); final CreditBasedClientHandler handler = new CreditBasedClientHandler(); - final EmbeddedChannel channel = new EmbeddedChannel(handler); + final EmbeddedChannel channel = spy(new EmbeddedChannel(handler)); - final RemoteInputChannel inputChannel = createRemoteInputChannel(mock(SingleInputGate.class)); + // Increase the credits to enqueue the input channels + inputChannel1.increaseCredit(1); + inputChannel2.increaseCredit(1); + handler.notifyCreditAvailable(inputChannel1); + handler.notifyCreditAvailable(inputChannel2); - // Enqueue the input channel - handler.notifyCreditAvailable(inputChannel); + channel.runPendingTasks(); + + // The two input channels should notify credits via writable channel + assertTrue(channel.isWritable()); + assertEquals(channel.readOutbound().getClass(), AddCredit.class); --- End diff -- how about checking the actual contents of the `AddCredit` message here, too? Also, shouldn't we receive two of those messages now? Something like this: ``` Object readFromOutbound = channel.readOutbound(); assertThat(readFromOutbound, instanceOf(AddCredit.class)); assertEquals(1, ((AddCredit) readFromOutbound).credit); readFromOutbound = channel.readOutbound(); assertThat(readFromOutbound, instanceOf(AddCredit.class)); assertEquals(1, ((AddCredit) readFromOutbound).credit); assertNull(channel.readOutbound()); ``` (also in general, it is `assertEquals(expected, actual)`, so the order of the parameters is wrong - not only here)
---