RobertIndie commented on code in PR #21183:
URL: https://github.com/apache/pulsar/pull/21183#discussion_r1361433477
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java:
##########
@@ -1011,10 +1016,114 @@ 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.
Review Comment:
```suggestion
// Verify the second consumer using a new connection will override
the consumer who using a stopped channel.
```
Consumer ?
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java:
##########
@@ -1011,10 +1016,114 @@ 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);
+ AtomicBoolean channel1MonitorStopped =
startChannelMonitorToHandleUserTask();
+
+ assertTrue(getResponse(channel2.channel, channel2.clientChannelHelper)
instanceof CommandSuccess);
+ assertEquals(topicRef.getSubscription(sName).getConsumers().size(), 1);
+
assertEquals(topicRef.getSubscription(sName).getConsumers().iterator().next().consumerName(),
cName2);
+ channel1MonitorStopped.set(true);
+
+ // cleanup.
+ channel.finish();
+ channel2.close();
+ }
+
+ @Test
+ public void
testHandleConsumerAfterClientChannelInactiveWithDisabledConnectionCheck()
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();
+ // Disabled connection check.
+ pulsar.getConfig().setConnectionLivenessCheckTimeoutMillis(-1);
+ 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);
+ AtomicBoolean channel1MonitorStopped =
startChannelMonitorToHandleUserTask();
+
+ Object responseOfConnection2 = getResponse(channel2.channel,
channel2.clientChannelHelper);
+ assertTrue(responseOfConnection2 instanceof CommandError);
+ assertTrue(((CommandError) responseOfConnection2).getMessage()
+ .contains("Exclusive consumer is already connected"));
+ assertEquals(topicRef.getSubscription(sName).getConsumers().size(), 1);
+
assertEquals(topicRef.getSubscription(sName).getConsumers().iterator().next().consumerName(),
cName1);
+ channel1MonitorStopped.set(true);
+
+ // cleanup.
+ channel.finish();
+ channel2.close();
+ // Reset configuration.
+ pulsar.getConfig().setConnectionLivenessCheckTimeoutMillis(5000);
+ }
+
+ private AtomicBoolean startChannelMonitorToHandleUserTask() {
Review Comment:
This seems like running a timer task. Maybe we can refactor it using the
timer to make it easier to understand.
--
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]
