poorbarcode commented on code in PR #21183:
URL: https://github.com/apache/pulsar/pull/21183#discussion_r1374147582
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java:
##########
@@ -1011,10 +1020,132 @@ public void
testHandleProducerAfterClientChannelInactive() throws Exception {
channel2.close();
}
+ @Test
+ public void testHandleConsumerAfterClientChannelInactive() throws
Exception {
+ final String tName = successTopicName;
+ final long consumerId = 1;
+ final MutableInt requestId = new MutableInt(1);
+ final String sName = successSubName;
+ final String cName1 = ConsumerName.generateRandomName();
+ final String cName2 = ConsumerName.generateRandomName();
+ resetChannel();
+ setChannelConnected();
+
+ // The producer register using the first connection.
+ ByteBuf cmdSubscribe1 = Commands.newSubscribe(tName, sName,
consumerId, requestId.incrementAndGet(),
+ SubType.Exclusive, 0, cName1, 0);
+ channel.writeInbound(cmdSubscribe1);
+ assertTrue(getResponse() instanceof CommandSuccess);
+ PersistentTopic topicRef = (PersistentTopic)
brokerService.getTopicReference(tName).get();
+ assertNotNull(topicRef);
+ assertNotNull(topicRef.getSubscription(sName).getConsumers());
+ assertEquals(topicRef.getSubscription(sName).getConsumers().size(), 1);
+
assertEquals(topicRef.getSubscription(sName).getConsumers().iterator().next().consumerName(),
cName1);
+
+ // Verify the second producer using a new connection will override the
consumer who using a stopped channel.
+ channelsStoppedAnswerHealthCheck.add(channel);
+ ClientChannel channel2 = new ClientChannel();
+ setChannelConnected(channel2.serverCnx);
+ ByteBuf cmdSubscribe2 = Commands.newSubscribe(tName, sName,
consumerId, requestId.incrementAndGet(),
+ CommandSubscribe.SubType.Exclusive, 0, cName2, 0);
+ channel2.channel.writeInbound(cmdSubscribe2);
+ BackGroundExecutor backGroundExecutor =
startBackgroundExecutorForEmbeddedChannel(channel);
+
+ assertTrue(getResponse(channel2.channel, channel2.clientChannelHelper)
instanceof CommandSuccess);
+ assertEquals(topicRef.getSubscription(sName).getConsumers().size(), 1);
+
assertEquals(topicRef.getSubscription(sName).getConsumers().iterator().next().consumerName(),
cName2);
+ backGroundExecutor.close();
+
+ // cleanup.
+ channel.finish();
+ channel2.close();
+ }
+
+ @Test
+ public void
testHandleConsumerAfterClientChannelInactiveWhenDisabledFeatureConnectionLivenessCheckTimeoutMillis()
Review Comment:
> This method name is too long
Renamed it.
> Could you add another method for the common logic?
If we add a comment method for these two methods, the args of the commont
method is too much, so it is not graceful. Since there are only three steps, we
can write them in each test method.
> And we can use DataProvider for it.
No, for these two test: one expected success, and another expected failed,
we should not merge them into one
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
