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/08/19 16:39:42 UTC
[GitHub] [pinot] sajjad-moradi opened a new pull request, #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
sajjad-moradi opened a new pull request, #9249:
URL: https://github.com/apache/pinot/pull/9249
Setting up consumption rate limit requires knowing the stream partition count so that each partition rate limit can be calculated based on the topic consumption rate limit config. Connecting to stream for fetching the partition count can time out or face any other issue. This PR handle that connection failure by using the previously fetch value.
--
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] PrachiKhobragade commented on a diff in pull request #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
PrachiKhobragade commented on code in PR #9249:
URL: https://github.com/apache/pinot/pull/9249#discussion_r950550273
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java:
##########
@@ -131,4 +144,22 @@ public void throttle(int numMsgs) {
return _rate;
}
}
+
+ @VisibleForTesting
+ @FunctionalInterface
+ interface PartitionCountFetcher {
+ Integer fetch(StreamConfig streamConfig);
+ }
+
+ @VisibleForTesting
+ static final PartitionCountFetcher DEFAULT_PARTITION_COUNT_FETCHER = streamConfig -> {
+ String clientId = streamConfig.getTopicName() + "-consumption.rate.manager";
+ StreamConsumerFactory factory = StreamConsumerFactoryProvider.create(streamConfig);
+ try (StreamMetadataProvider streamMetadataProvider = factory.createStreamMetadataProvider(clientId)) {
+ return streamMetadataProvider.fetchPartitionCount(/*maxWaitTimeMs*/10_000);
Review Comment:
we need to honor this in the implementation
--
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] sajjad-moradi commented on a diff in pull request #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on code in PR #9249:
URL: https://github.com/apache/pinot/pull/9249#discussion_r950491202
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java:
##########
@@ -85,17 +88,27 @@ public ConsumptionRateLimiter createRateLimiter(StreamConfig streamConfig, Strin
return new RateLimiterImpl(partitionRateLimit);
}
- private static LoadingCache<StreamConfig, Integer> buildCache() {
- return CacheBuilder.newBuilder().expireAfterWrite(CACHE_ENTRY_EXPIRATION_TIME_IN_MINUTES, TimeUnit.MINUTES)
+ @VisibleForTesting
+ static LoadingCache<StreamConfig, Integer> buildCache(PartitionCountFetcher partitionCountFetcher,
+ long duration, TimeUnit unit) {
+ return CacheBuilder.newBuilder().refreshAfterWrite(duration, unit)
.build(new CacheLoader<StreamConfig, Integer>() {
@Override
- public Integer load(StreamConfig streamConfig)
+ public Integer load(StreamConfig key)
Review Comment:
Because that's not enough to create a consumer. `StreamConsumerFactory` requires the whole streamConfig object.
--
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] sajjad-moradi commented on a diff in pull request #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on code in PR #9249:
URL: https://github.com/apache/pinot/pull/9249#discussion_r950500727
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java:
##########
@@ -131,4 +144,22 @@ public void throttle(int numMsgs) {
return _rate;
}
}
+
+ @VisibleForTesting
+ @FunctionalInterface
+ interface PartitionCountFetcher {
+ Integer fetch(StreamConfig streamConfig);
+ }
+
+ @VisibleForTesting
+ static final PartitionCountFetcher DEFAULT_PARTITION_COUNT_FETCHER = streamConfig -> {
+ String clientId = streamConfig.getTopicName() + "-consumption.rate.manager";
+ StreamConsumerFactory factory = StreamConsumerFactoryProvider.create(streamConfig);
+ try (StreamMetadataProvider streamMetadataProvider = factory.createStreamMetadataProvider(clientId)) {
+ return streamMetadataProvider.fetchPartitionCount(/*maxWaitTimeMs*/10_000);
+ } catch (Exception e) {
+ LOGGER.warn("Error fetching metadata for topic " + streamConfig.getTopicName(), e);
Review Comment:
I checked the server side and there's none. I agree we should add one metric for when there's a stream connection failures, but IMO that should be a separate PR.
--
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] sajjad-moradi commented on pull request #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
sajjad-moradi commented on PR #9249:
URL: https://github.com/apache/pinot/pull/9249#issuecomment-1221184780
> We had discussed about a metadata cache. is that something that is coming in another PR (and this one is to get us through the immediate problem)?
>
> thanks
Yes that will be another PR. This PR is just to make sure we properly handle stream connection exceptions.
--
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] snleee commented on a diff in pull request #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
snleee commented on code in PR #9249:
URL: https://github.com/apache/pinot/pull/9249#discussion_r950487153
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java:
##########
@@ -131,4 +144,22 @@ public void throttle(int numMsgs) {
return _rate;
}
}
+
+ @VisibleForTesting
+ @FunctionalInterface
+ interface PartitionCountFetcher {
+ Integer fetch(StreamConfig streamConfig);
+ }
+
+ @VisibleForTesting
+ static final PartitionCountFetcher DEFAULT_PARTITION_COUNT_FETCHER = streamConfig -> {
+ String clientId = streamConfig.getTopicName() + "-consumption.rate.manager";
+ StreamConsumerFactory factory = StreamConsumerFactoryProvider.create(streamConfig);
+ try (StreamMetadataProvider streamMetadataProvider = factory.createStreamMetadataProvider(clientId)) {
+ return streamMetadataProvider.fetchPartitionCount(/*maxWaitTimeMs*/10_000);
+ } catch (Exception e) {
+ LOGGER.warn("Error fetching metadata for topic " + streamConfig.getTopicName(), e);
Review Comment:
Do we have any metrics for Kafka connection failure on the server side consuming segment or the controller side?
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java:
##########
@@ -85,17 +88,27 @@ public ConsumptionRateLimiter createRateLimiter(StreamConfig streamConfig, Strin
return new RateLimiterImpl(partitionRateLimit);
}
- private static LoadingCache<StreamConfig, Integer> buildCache() {
- return CacheBuilder.newBuilder().expireAfterWrite(CACHE_ENTRY_EXPIRATION_TIME_IN_MINUTES, TimeUnit.MINUTES)
+ @VisibleForTesting
+ static LoadingCache<StreamConfig, Integer> buildCache(PartitionCountFetcher partitionCountFetcher,
+ long duration, TimeUnit unit) {
+ return CacheBuilder.newBuilder().refreshAfterWrite(duration, unit)
.build(new CacheLoader<StreamConfig, Integer>() {
@Override
- public Integer load(StreamConfig streamConfig)
+ public Integer load(StreamConfig key)
Review Comment:
Why don't we use the table name as the key instead of the entire `streamConfig`?
--
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 #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #9249:
URL: https://github.com/apache/pinot/pull/9249#issuecomment-1220908246
# [Codecov](https://codecov.io/gh/apache/pinot/pull/9249?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 [#9249](https://codecov.io/gh/apache/pinot/pull/9249?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (c3d7a22) into [master](https://codecov.io/gh/apache/pinot/commit/6c5d85ef7eeffcd4cddd639865b81764fb04dbf1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (6c5d85e) will **decrease** coverage by `2.81%`.
> The diff coverage is `53.33%`.
> :exclamation: Current head c3d7a22 differs from pull request most recent head e6eaa85. Consider uploading reports for the commit e6eaa85 to get more accurate results
```diff
@@ Coverage Diff @@
## master #9249 +/- ##
============================================
- Coverage 69.91% 67.10% -2.82%
- Complexity 4764 4815 +51
============================================
Files 1858 1381 -477
Lines 99187 71954 -27233
Branches 15087 11521 -3566
============================================
- Hits 69346 48282 -21064
+ Misses 24930 20169 -4761
+ Partials 4911 3503 -1408
```
| Flag | Coverage Δ | |
|---|---|---|
| integration1 | `?` | |
| integration2 | `?` | |
| unittests1 | `67.10% <53.33%> (+0.01%)` | :arrow_up: |
| unittests2 | `?` | |
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/9249?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...nager/realtime/RealtimeConsumptionRateManager.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvUmVhbHRpbWVDb25zdW1wdGlvblJhdGVNYW5hZ2VyLmphdmE=) | `69.56% <53.33%> (-6.12%)` | :arrow_down: |
| [...va/org/apache/pinot/core/routing/RoutingTable.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9yb3V0aW5nL1JvdXRpbmdUYWJsZS5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...va/org/apache/pinot/common/config/NettyConfig.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vY29uZmlnL05ldHR5Q29uZmlnLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...a/org/apache/pinot/common/metrics/MinionMeter.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9NaW5pb25NZXRlci5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...g/apache/pinot/common/metrics/ControllerMeter.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9Db250cm9sbGVyTWV0ZXIuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [.../apache/pinot/common/metrics/BrokerQueryPhase.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9Ccm9rZXJRdWVyeVBoYXNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [.../apache/pinot/common/metrics/MinionQueryPhase.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9NaW5pb25RdWVyeVBoYXNlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ache/pinot/server/access/AccessControlFactory.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VydmVyL2FjY2Vzcy9BY2Nlc3NDb250cm9sRmFjdG9yeS5qYXZh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...he/pinot/common/messages/SegmentReloadMessage.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWVzc2FnZXMvU2VnbWVudFJlbG9hZE1lc3NhZ2UuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...he/pinot/common/messages/TableDeletionMessage.java](https://codecov.io/gh/apache/pinot/pull/9249/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-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWVzc2FnZXMvVGFibGVEZWxldGlvbk1lc3NhZ2UuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [714 more](https://codecov.io/gh/apache/pinot/pull/9249/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) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?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] snleee commented on a diff in pull request #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
snleee commented on code in PR #9249:
URL: https://github.com/apache/pinot/pull/9249#discussion_r950515885
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeConsumptionRateManager.java:
##########
@@ -85,17 +88,27 @@ public ConsumptionRateLimiter createRateLimiter(StreamConfig streamConfig, Strin
return new RateLimiterImpl(partitionRateLimit);
}
- private static LoadingCache<StreamConfig, Integer> buildCache() {
- return CacheBuilder.newBuilder().expireAfterWrite(CACHE_ENTRY_EXPIRATION_TIME_IN_MINUTES, TimeUnit.MINUTES)
+ @VisibleForTesting
+ static LoadingCache<StreamConfig, Integer> buildCache(PartitionCountFetcher partitionCountFetcher,
+ long duration, TimeUnit unit) {
+ return CacheBuilder.newBuilder().refreshAfterWrite(duration, unit)
.build(new CacheLoader<StreamConfig, Integer>() {
@Override
- public Integer load(StreamConfig streamConfig)
+ public Integer load(StreamConfig key)
Review Comment:
I see. It looks that the Guava's LoaderCache doesn't allow us to pass the extra parameter than the key object.
--
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] sajjad-moradi merged pull request #9249: Gracefully handle problem in fetching stream partition count for consumption rate limiter
Posted by GitBox <gi...@apache.org>.
sajjad-moradi merged PR #9249:
URL: https://github.com/apache/pinot/pull/9249
--
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