You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "BewareMyPower (via GitHub)" <gi...@apache.org> on 2024/01/22 13:50:22 UTC
[PR] [fix][client] Fix seeking by timestamp does not work for multi-topics [pulsar]
BewareMyPower opened a new pull request, #21945:
URL: https://github.com/apache/pulsar/pull/21945
### Motivation
The multi-topics consumer supports seeking all internal consumers when accepting a timestamp as the argument. However, the multi-topics consumer could still read from earliest after seeking to a specific position. It's because before an internal consumer completes a seek operation, it could still receive duplicated messages.
### Modifications
For a multi-topics consumer, before the seek operation is done, do not call `messageReceived` method. Once the seek operation is done, collect all messages from internal consumers and call `messageReceived` method for each one.
Add `testSeekByTimestamp` to verify `receive`, `receiveAsync` and message listener all work for seeking by timestamp on a multi-topics consumer.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository:
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#discussion_r1469598386
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -775,25 +776,44 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
);
}
- final CompletableFuture<Void> seekFuture;
if (internalConsumer == null) {
- List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
- consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(messageId)));
- seekFuture = FutureUtil.waitForAll(futures);
+ return seekAllAsync(consumer -> consumer.seekAsync(messageId));
} else {
- seekFuture = internalConsumer.seekAsync(messageId);
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ internalConsumer.seekAsync(messageId).whenComplete((__, e) -> afterSeek(future, e));
+ return future;
}
+ }
+ @Override
+ public CompletableFuture<Void> seekAsync(long timestamp) {
+ return seekAllAsync(consumer -> consumer.seekAsync(timestamp));
+ }
+
+ private CompletableFuture<Void> seekAllAsync(Function<ConsumerImpl<T>, CompletableFuture<Void>> seekFunc) {
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ FutureUtil.waitForAll(consumers.values().stream().map(seekFunc).collect(Collectors.toList()))
+ .whenComplete((__, e) -> afterSeek(future, e));
+ return future;
+ }
+
+ private void beforeSeek() {
+ duringSeek = true;
Review Comment:
The internal consumer's `seekAsync` will fail so that this multi-topics consumer's `seekAsync` will fail with the same exception.
https://github.com/apache/pulsar/blob/8a18043585c49624e02b875b921336a81a9ec577/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L2261
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#discussion_r1469448666
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -775,25 +776,44 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
);
}
- final CompletableFuture<Void> seekFuture;
if (internalConsumer == null) {
- List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
- consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(messageId)));
- seekFuture = FutureUtil.waitForAll(futures);
+ return seekAllAsync(consumer -> consumer.seekAsync(messageId));
} else {
- seekFuture = internalConsumer.seekAsync(messageId);
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ internalConsumer.seekAsync(messageId).whenComplete((__, e) -> afterSeek(future, e));
+ return future;
}
+ }
+ @Override
+ public CompletableFuture<Void> seekAsync(long timestamp) {
+ return seekAllAsync(consumer -> consumer.seekAsync(timestamp));
+ }
+
+ private CompletableFuture<Void> seekAllAsync(Function<ConsumerImpl<T>, CompletableFuture<Void>> seekFunc) {
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ FutureUtil.waitForAll(consumers.values().stream().map(seekFunc).collect(Collectors.toList()))
+ .whenComplete((__, e) -> afterSeek(future, e));
+ return future;
+ }
+
+ private void beforeSeek() {
Review Comment:
Do we need to call `resetIncomingMessageSize();` in this method?
I noticed we have called at [here](https://github.com/apache/pulsar/pull/21945/files#diff-b1cc861088865a65da05597a90794f9910e0a613fb6eb24e343cec5b5ec7da47L755) but it will be removed in this PR.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -252,7 +258,7 @@ private void receiveMessageFromConsumer(ConsumerImpl<T> consumer, boolean batchR
}
// Process the message, add to the queue and trigger listener or async callback
messages.forEach(msg -> {
- if (isValidConsumerEpoch((MessageImpl<T>) msg)) {
+ if (isValidConsumerEpoch((MessageImpl<T>) msg) && !duringSeek) {
Review Comment:
We'd better add a DEBUG level log here to know if the messages were skipped during the seek operation. It will help us troubleshoot some problems related to the message that was not received from the user's perspective.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -775,25 +776,44 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
);
}
- final CompletableFuture<Void> seekFuture;
if (internalConsumer == null) {
- List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
- consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(messageId)));
- seekFuture = FutureUtil.waitForAll(futures);
+ return seekAllAsync(consumer -> consumer.seekAsync(messageId));
} else {
- seekFuture = internalConsumer.seekAsync(messageId);
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ internalConsumer.seekAsync(messageId).whenComplete((__, e) -> afterSeek(future, e));
+ return future;
Review Comment:
It looks like we can also use `seekAllAsync()` method by changing the method signature to `seekAllAsync(List<Consumer> consumers, (Function<ConsumerImpl<T>, CompletableFuture<Void>> seekFunc)`?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -775,25 +776,44 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
);
}
- final CompletableFuture<Void> seekFuture;
if (internalConsumer == null) {
- List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
- consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(messageId)));
- seekFuture = FutureUtil.waitForAll(futures);
+ return seekAllAsync(consumer -> consumer.seekAsync(messageId));
} else {
- seekFuture = internalConsumer.seekAsync(messageId);
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ internalConsumer.seekAsync(messageId).whenComplete((__, e) -> afterSeek(future, e));
+ return future;
}
+ }
+ @Override
+ public CompletableFuture<Void> seekAsync(long timestamp) {
+ return seekAllAsync(consumer -> consumer.seekAsync(timestamp));
+ }
+
+ private CompletableFuture<Void> seekAllAsync(Function<ConsumerImpl<T>, CompletableFuture<Void>> seekFunc) {
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ FutureUtil.waitForAll(consumers.values().stream().map(seekFunc).collect(Collectors.toList()))
+ .whenComplete((__, e) -> afterSeek(future, e));
+ return future;
+ }
+
+ private void beforeSeek() {
Review Comment:
If we need to add it back, we'd better also add a test to cover this case.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#issuecomment-1917399740
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: `5 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`17bb322`)](https://app.codecov.io/gh/apache/pulsar/commit/17bb3227536f891d4f2ef234da596bdb5bb46bbb?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 73.59% compared to head [(`1ee4441`)](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 73.67%.
> Report is 25 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #21945 +/- ##
============================================
+ Coverage 73.59% 73.67% +0.07%
- Complexity 32417 32489 +72
============================================
Files 1861 1863 +2
Lines 138678 138784 +106
Branches 15188 15207 +19
============================================
+ Hits 102060 102245 +185
+ Misses 28715 28651 -64
+ Partials 7903 7888 -15
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/21945/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/21945/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.08% <12.50%> (-0.05%)` | :arrow_down: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/21945/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.65% <12.50%> (-0.03%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/21945/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `72.94% <84.37%> (+0.07%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...n/java/org/apache/pulsar/broker/PulsarService.java](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9QdWxzYXJTZXJ2aWNlLmphdmE=) | `83.06% <100.00%> (+0.74%)` | :arrow_up: |
| [...ache/pulsar/client/admin/internal/BrokersImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC1hZG1pbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9hZG1pbi9pbnRlcm5hbC9Ccm9rZXJzSW1wbC5qYXZh) | `87.50% <100.00%> (ø)` | |
| [...g/apache/pulsar/broker/admin/impl/BrokersBase.java](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9hZG1pbi9pbXBsL0Jyb2tlcnNCYXNlLmphdmE=) | `70.86% <50.00%> (ø)` | |
| [...he/pulsar/client/impl/MultiTopicsConsumerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/21945?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL011bHRpVG9waWNzQ29uc3VtZXJJbXBsLmphdmE=) | `78.41% <84.61%> (+0.31%)` | :arrow_up: |
... and [113 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/21945/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix seeking by timestamp does not work for multi-topics [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#issuecomment-1905435990
This issue should also exist for other `seekAsync` overrides, I will refactor the seek logic soon.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- closed pull request #21945: [fix][client] Fix multi-topics consumer could receive old messages after seek
URL: https://github.com/apache/pulsar/pull/21945
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- merged PR #21945:
URL: https://github.com/apache/pulsar/pull/21945
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#discussion_r1469320442
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -775,25 +776,44 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
);
}
- final CompletableFuture<Void> seekFuture;
if (internalConsumer == null) {
- List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
- consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(messageId)));
- seekFuture = FutureUtil.waitForAll(futures);
+ return seekAllAsync(consumer -> consumer.seekAsync(messageId));
} else {
- seekFuture = internalConsumer.seekAsync(messageId);
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ internalConsumer.seekAsync(messageId).whenComplete((__, e) -> afterSeek(future, e));
+ return future;
}
+ }
+ @Override
+ public CompletableFuture<Void> seekAsync(long timestamp) {
+ return seekAllAsync(consumer -> consumer.seekAsync(timestamp));
+ }
+
+ private CompletableFuture<Void> seekAllAsync(Function<ConsumerImpl<T>, CompletableFuture<Void>> seekFunc) {
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ FutureUtil.waitForAll(consumers.values().stream().map(seekFunc).collect(Collectors.toList()))
+ .whenComplete((__, e) -> afterSeek(future, e));
+ return future;
+ }
+
+ private void beforeSeek() {
+ duringSeek = true;
Review Comment:
What if we have the second seek happened concurrently while handling the first seek? There seems to be a chance that the consumer still be able to receive messages during the second concurrent seek. Should we isolate them?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#discussion_r1469606966
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -775,25 +776,44 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
);
}
- final CompletableFuture<Void> seekFuture;
if (internalConsumer == null) {
- List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
- consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(messageId)));
- seekFuture = FutureUtil.waitForAll(futures);
+ return seekAllAsync(consumer -> consumer.seekAsync(messageId));
} else {
- seekFuture = internalConsumer.seekAsync(messageId);
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ internalConsumer.seekAsync(messageId).whenComplete((__, e) -> afterSeek(future, e));
+ return future;
Review Comment:
Yes (the 1st parameter should be `Collection<Consumer>`)
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#discussion_r1469593762
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -775,25 +776,44 @@ public CompletableFuture<Void> seekAsync(MessageId messageId) {
);
}
- final CompletableFuture<Void> seekFuture;
if (internalConsumer == null) {
- List<CompletableFuture<Void>> futures = new ArrayList<>(consumers.size());
- consumers.values().forEach(consumerImpl -> futures.add(consumerImpl.seekAsync(messageId)));
- seekFuture = FutureUtil.waitForAll(futures);
+ return seekAllAsync(consumer -> consumer.seekAsync(messageId));
} else {
- seekFuture = internalConsumer.seekAsync(messageId);
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ internalConsumer.seekAsync(messageId).whenComplete((__, e) -> afterSeek(future, e));
+ return future;
}
+ }
+ @Override
+ public CompletableFuture<Void> seekAsync(long timestamp) {
+ return seekAllAsync(consumer -> consumer.seekAsync(timestamp));
+ }
+
+ private CompletableFuture<Void> seekAllAsync(Function<ConsumerImpl<T>, CompletableFuture<Void>> seekFunc) {
+ beforeSeek();
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ FutureUtil.waitForAll(consumers.values().stream().map(seekFunc).collect(Collectors.toList()))
+ .whenComplete((__, e) -> afterSeek(future, e));
+ return future;
+ }
+
+ private void beforeSeek() {
Review Comment:
`resetIncomingMessageSize()` is called in `clearIncomingMessages()`, I will add more assertions to protect it.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#discussion_r1469616942
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -252,7 +258,7 @@ private void receiveMessageFromConsumer(ConsumerImpl<T> consumer, boolean batchR
}
// Process the message, add to the queue and trigger listener or async callback
messages.forEach(msg -> {
- if (isValidConsumerEpoch((MessageImpl<T>) msg)) {
+ if (isValidConsumerEpoch((MessageImpl<T>) msg) && !duringSeek) {
Review Comment:
I think INFO log would be better. It's a rare case to come here because the check is already done at the beginning of `receiveMessageFromConsumer`. Even if it came here, the maximum size of `messages` is 1 or `batchMaxMessages` (by default 1000).
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][client] Fix multi-topics consumer could receive old messages after seek [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #21945:
URL: https://github.com/apache/pulsar/pull/21945#issuecomment-1914732578
@codelipenghui I have addressed your comments, PTAL again.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org