Enhance PulsarConsumerTest

[Jackie-Jiang]

• Reduce fetch timeout so that it needs to fetch multiple batches and test behavior
• Fix producer to properly produce batched messages

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 62.24%. Comparing base (59551e4) to head (e4b4c28).
Report is 326 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #12948      +/-   ##
============================================
+ Coverage     61.75%   62.24%   +0.49%     
+ Complexity      207      198       -9     
============================================
  Files          2436     2502      +66     
  Lines        133233   136466    +3233     
  Branches      20636    21120     +484     
============================================
+ Hits          82274    84942    +2668     
- Misses        44911    45257     +346     
- Partials       6048     6267     +219     

Flag	Coverage Δ
custom-integration1	<0.01% <ø> (-0.01%)	⬇️
integration	<0.01% <ø> (-0.01%)	⬇️
integration1	<0.01% <ø> (-0.01%)	⬇️
integration2	0.00% <ø> (ø)
java-11	62.20% <ø> (+0.49%)	⬆️
java-21	62.11% <ø> (+0.49%)	⬆️
skip-bytebuffers-false	62.23% <ø> (+0.48%)	⬆️
skip-bytebuffers-true	62.08% <ø> (+34.35%)	⬆️
temurin	62.24% <ø> (+0.49%)	⬆️
unittests	62.23% <ø> (+0.49%)	⬆️
unittests1	46.70% <ø> (-0.19%)	⬇️
unittests2	28.02% <ø> (+0.29%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[Jackie-Jiang]

Got this failure:

Error:  org.apache.pinot.plugin.stream.pulsar.PulsarConsumerTest.testPartitionLevelConsumerBatchMessages -- Time elapsed: 9.106 s <<< FAILURE!
java.lang.AssertionError: expected [sample_msg_100] but found [sample_msg_0]
	at org.testng.Assert.fail(Assert.java:111)
	at org.testng.Assert.failNotEquals(Assert.java:1578)
	at org.testng.Assert.assertEqualsImpl(Assert.java:150)
	at org.testng.Assert.assertEquals(Assert.java:132)
	at org.testng.Assert.assertEquals(Assert.java:656)
	at org.testng.Assert.assertEquals(Assert.java:666)
	at org.apache.pinot.plugin.stream.pulsar.PulsarConsumerTest.verifyMessage(PulsarConsumerTest.java:219)
	at org.apache.pinot.plugin.stream.pulsar.PulsarConsumerTest.testConsumer(PulsarConsumerTest.java:208)
	at org.apache.pinot.plugin.stream.pulsar.PulsarConsumerTest.testPartitionLevelConsumerBatchMessages(PulsarConsumerTest.java:251)

Seems like the first 100 messages are somehow reproduced. This aligns with the previous failures

[Jackie-Jiang]

Turns out we were not really publish messages in batches. Fixed the producer to publish in batches and verifies that.

[Jackie-Jiang]

Fix #8537

[klsince]

👍