[BUG] Tests fail with mismatched messageIds against pulsar 2.5.0

[addisonj]

Describe the bug
When changing this code to run against pulsar 2.5.0, the tests fail. The main error from the tests looks like so:

java.lang.Exception: java.lang.IllegalStateException: Potential Data Loss in reading persistent://public/default/topic-53-partition-4: intended to start at 96:0:4:303, actually we get 96:0:4:0Some data may have been lost because they are not available in Pulsar any more; either the
 data was aged out by Pulsar or the topic may have been deleted before all the data in the
 topic was processed. If you don't want your streaming query to fail on such cases, set the
 source option "failOnDataLoss" to "false".

As can be seen, the messageIds almost match, just the batchIndex is getting lost. It would appear that when we are creating new readers, the broker is responding with a subscription from the beginning of the batch, not at the requested batchIndex.

It isn't clear if this is intended behavior in 2.5.0 or if this would is a regression in Pulsar. If this is intended, then this code needs to change to either skip or call seek on the reader again to a more specific position

To Reproduce
Steps to reproduce the behavior:

1. Change the pulsar version to 2.5.0 and the test framework to 2.5.1
2. Run the tests
3. Observe the error

Expected behavior
Tests should pass

[addisonj]

I may be wrong, but I think this might be the commit where it changed in pulsar: apache/pulsar#4331

[sijie]

@addisonj I think the regression is in 2.5.0. #4331 has fixed the problem.

[syhily]

Close this issue for outdated.