Fixed BrokerClientIntegrationTest.testUnsupportedBatchMessageConsumer
### Motivation
Fixes The second consumer was being created with `Exclusive` type and was actually failing to connect.
@@ -310,15 +310,23 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
final String topicName = "persistent://my-property/my-ns/my-topic1";
final String subscriptionName = "my-subscriber-name" + subType;
- ConsumerImpl<byte[]> consumer1 = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(topicName)
- .subscriptionName(subscriptionName).subscriptionType(subType).subscribe();
+ ConsumerImpl<byte[]> consumer1 = (ConsumerImpl<byte[]>) pulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(subType)
+ .subscribe();
final int numMessagesPerBatch = 10;
- Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
- Producer<byte[]> batchProducer = pulsarClient.newProducer().topic(topicName).enableBatching(true)
+ Producer<byte[]> producer = pulsarClient.newProducer()
+ .topic(topicName)
+ .enableBatching(false)
+ .create();
+ Producer<byte[]> batchProducer = pulsarClient.newProducer()
+ .topic(topicName).enableBatching(true)
.batchingMaxPublishDelay(Long.MAX_VALUE, TimeUnit.SECONDS)
- .batchingMaxMessages(numMessagesPerBatch).create();
+ .batchingMaxMessages(numMessagesPerBatch)
+ .create();
// update consumer's version to incompatible batch-message version = Version.V3
Topic topic = pulsar.getBrokerService().getOrCreateTopic(topicName).get();
@@ -357,12 +365,15 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
batchProducer.flush();
// consumer should have not received any message as it should have been disconnected
- msg = consumer1.receive(2, TimeUnit.SECONDS);
+ msg = consumer1.receive(100, TimeUnit.MILLISECONDS);
assertNull(msg);
// subscribe consumer2 with supporting batch version
PulsarClient newPulsarClient = newPulsarClient(lookupUrl.toString(), 0); // Creates new client connection
- Consumer<byte[]> consumer2 = newPulsarClient.newConsumer().topic(topicName).subscriptionName(subscriptionName)
+ Consumer<byte[]> consumer2 = newPulsarClient.newConsumer()
+ .topic(topicName)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(subType)
.subscribe();
messageSet.clear();
