You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2022/04/13 21:47:34 UTC
[GitHub] [pinot] walterddr opened a new pull request, #8538: fix pulsar test instability
walterddr opened a new pull request, #8538:
URL: https://github.com/apache/pinot/pull/8538
This fixes #8537
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] KKcorps commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
KKcorps commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r851301660
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
This method was added to handle the scenarios where messages are published in batches which actually changes the format of MessageIds in Pulsar.
The only way to publish in batch is to enable async producer. In sync mode, all messages are flushed as soon as they are received.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] KKcorps commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
KKcorps commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r851344097
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
@xiangfu0 your suggestion also works. I should actually add a method like `waitForAllRecordsToFlush` and only then exit the setup.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] xiangfu0 commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
xiangfu0 commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r849948647
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
do we know why this call may give fewer records? Is it a bug or something like messages not flushed entirely and available to consume or some messages are actually missing
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] walterddr closed pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
walterddr closed pull request #8538: fix pulsar test instability
URL: https://github.com/apache/pinot/pull/8538
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] codecov-commenter commented on pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #8538:
URL: https://github.com/apache/pinot/pull/8538#issuecomment-1100700229
# [Codecov](https://codecov.io/gh/apache/pinot/pull/8538?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#8538](https://codecov.io/gh/apache/pinot/pull/8538?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (4e23c04) into [master](https://codecov.io/gh/apache/pinot/commit/2704d887324014817097e609b1e7d794a898ea99?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (2704d88) will **decrease** coverage by `3.78%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #8538 +/- ##
==========================================
- Coverage 29.52% 25.73% -3.79%
==========================================
Files 1674 1674
Lines 87872 87872
Branches 13313 13313
==========================================
- Hits 25942 22615 -3327
- Misses 59551 63102 +3551
+ Partials 2379 2155 -224
```
| Flag | Coverage Δ | |
|---|---|---|
| integration1 | `?` | |
| integration2 | `25.73% <ø> (-0.01%)` | :arrow_down: |
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=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/pinot/pull/8538?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...g/apache/pinot/server/api/resources/ErrorInfo.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3Qtc2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zZXJ2ZXIvYXBpL3Jlc291cmNlcy9FcnJvckluZm8uamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...apache/pinot/common/lineage/LineageEntryState.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbGluZWFnZS9MaW5lYWdlRW50cnlTdGF0ZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...pinot/minion/exception/TaskCancelledException.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtbWluaW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9taW5pb24vZXhjZXB0aW9uL1Rhc2tDYW5jZWxsZWRFeGNlcHRpb24uamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ker/failuredetector/ConnectionFailureDetector.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtYnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9icm9rZXIvZmFpbHVyZWRldGVjdG9yL0Nvbm5lY3Rpb25GYWlsdXJlRGV0ZWN0b3IuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...minion/tasks/mergerollup/MergeRollupTaskUtils.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vyb2xsdXAvTWVyZ2VSb2xsdXBUYXNrVXRpbHMuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ion/tasks/mergerollup/MergeRollupTaskExecutor.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvbWVyZ2Vyb2xsdXAvTWVyZ2VSb2xsdXBUYXNrRXhlY3V0b3IuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...nverttorawindex/ConvertToRawIndexTaskExecutor.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvY29udmVydHRvcmF3aW5kZXgvQ29udmVydFRvUmF3SW5kZXhUYXNrRXhlY3V0b3IuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...e/pinot/common/minion/MergeRollupTaskMetadata.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWluaW9uL01lcmdlUm9sbHVwVGFza01ldGFkYXRhLmphdmE=) | `0.00% <0.00%> (-94.74%)` | :arrow_down: |
| [...rg/apache/pinot/common/lineage/SegmentLineage.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbGluZWFnZS9TZWdtZW50TGluZWFnZS5qYXZh) | `0.00% <0.00%> (-91.31%)` | :arrow_down: |
| [...ache/pinot/common/lineage/SegmentLineageUtils.java](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbGluZWFnZS9TZWdtZW50TGluZWFnZVV0aWxzLmphdmE=) | `11.11% <0.00%> (-88.89%)` | :arrow_down: |
| ... and [191 more](https://codecov.io/gh/apache/pinot/pull/8538/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/pinot/pull/8538?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/pinot/pull/8538?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [2704d88...4e23c04](https://codecov.io/gh/apache/pinot/pull/8538?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] walterddr commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
walterddr commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r849987785
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
I saw the publishRecordBatch was changed in #8017 - any chance @KKcorps can give some insights?
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] KKcorps commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
KKcorps commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r851528375
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
There is another approach - https://github.com/apache/pinot/pull/8554
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] walterddr commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
walterddr commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r849972145
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
I dunno why this fixes the bug. My guess is that pulsar flush is async but I am not sure.
GitHub Actions sometimes catch interesting bug b/c the machines are much less powerful
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] walterddr commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
walterddr commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r851349057
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
according to the documentation https://pulsar.apache.org/api/client/org/apache/pulsar/client/api/Producer.html
- producer.flush() should wait until all the message finish publishing before returning (or throw exception); thus we can't check on the producer side
- only way to check and wait for all records reaches pulsar for consumption is to actually consume -- which is exactly what this test is doing.
so I guess we should do is wrap the test with
```
TestUtils.waitForCondition(aVoid -> {
// consume and assert
}, timeoutMs, "")
```
sounds good?
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] KKcorps commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
KKcorps commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r851344097
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
@xiangfu0 your suggestion also works. I can actually add a method like `waitForAllRecordsToFlush` and only then exit the setup.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] walterddr commented on pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
walterddr commented on PR #8538:
URL: https://github.com/apache/pinot/pull/8538#issuecomment-1100700893
This doesn't work. It seems like not a flakiness due to delay async produce.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] xiangfu0 commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
xiangfu0 commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r849952159
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
Shall we actually run a consumer to consume exact same number of messages then exit in setup() method?
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] KKcorps commented on a diff in pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
KKcorps commented on code in PR #8538:
URL: https://github.com/apache/pinot/pull/8538#discussion_r851528375
##########
pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java:
##########
@@ -273,13 +269,31 @@ public void testPartitionLevelConsumerBatchMessages()
}
}
- public MessageId getMessageIdForPartitionAndIndex(int partitionNum, int index) {
+ private static MessageBatch consumeMessageBatch(PartitionGroupConsumer consumer, MessageId startMsgId,
+ MessageId endMsgId, int expectedMsgCount)
+ throws TimeoutException {
+ int retryCount = 0;
+ MessageBatch messageBatch = null;
+ while (retryCount < DEFAULT_RETRY_COUNT) {
+ retryCount++;
+ messageBatch = consumer.fetchMessages(new MessageIdStreamOffset(startMsgId),
Review Comment:
There is another approach to just get topic stats. Since we are not producing messages in topics in any other method, it works for us.
I have implemented it here - https://github.com/apache/pinot/pull/8554
once this this PR is merged, I'll pull changes into my PR and we can merge that as well.
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] walterddr commented on pull request #8538: fix pulsar test instability
Posted by GitBox <gi...@apache.org>.
walterddr commented on PR #8538:
URL: https://github.com/apache/pinot/pull/8538#issuecomment-1100434001
it still fails. I wonder if theres' anything wrong with the consumer itself that it didn't pull enough data out.
alternatively I can still make the wait for condition checker in the test function too so that we can at least get better against the flakiness
--
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@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org