You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "Abhishek01911 (via GitHub)" <gi...@apache.org> on 2025/10/27 17:04:35 UTC
[PR] Fix: Handle kafka topic not found exception in fetchPartitionCount fu… [pinot]
Abhishek01911 opened a new pull request, #17088:
URL: https://github.com/apache/pinot/pull/17088
Summary
Resolves Issue [17045](https://github.com/apache/pinot/issues/17045)
This PR fixes an issue where an exception in fetchPartitionCount, often caused by a deleted topic, would halt ingestion for an entire multi-topic table. By catching this exception, the failure is now isolated. Partitions for the deleted topic will get stuck in a "consuming" state, but ingestion from all other valid topics will continue unaffected.
Testing
Tested on a pinot table. Validated single incorrect topic doesn't stop table ingestion completely
[
![Uploading Screenshot 2025-10-27 at 10.03.23 AM.png…]()
](url)
--
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
Re: [PR] Fix: Handle kafka topic not found exception in fetchPartitionCount fu… [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #17088:
URL: https://github.com/apache/pinot/pull/17088#discussion_r2470807412
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java:
##########
@@ -86,7 +87,24 @@ StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria offsetCriteri
default List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
throws IOException, TimeoutException {
- int partitionCount = fetchPartitionCount(timeoutMillis);
+
+ int partitionCount;
+ try {
+ partitionCount = fetchPartitionCount(timeoutMillis);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to fetch partition count for stream config: {}. Skipping stream and using"
+ + "existing partitions only. Error: {}", streamConfig.getTopicName(), e.getMessage(), e);
+ // Return only the existing partition groups if we can't fetch partition count
+ List<PartitionGroupMetadata> existingPartitionGroupMetadataList =
+ new ArrayList<>(partitionGroupConsumptionStatuses.size());
+ for (PartitionGroupConsumptionStatus currentPartitionGroupConsumptionStatus : partitionGroupConsumptionStatuses) {
+ existingPartitionGroupMetadataList.add(
+ new PartitionGroupMetadata(currentPartitionGroupConsumptionStatus.getStreamPartitionGroupId(),
+ currentPartitionGroupConsumptionStatus.getEndOffset()));
Review Comment:
How is this `partitionGroupConsumptionStatuses` updated?
Is it possible this `endOffset` is lower than the previous commit segment end offset?
--
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
Re: [PR] Fix: Handle kafka topic not found exception in fetchPartitionCount fu… [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17088:
URL: https://github.com/apache/pinot/pull/17088#discussion_r2535751654
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java:
##########
@@ -86,7 +87,26 @@ StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria offsetCriteri
default List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
throws IOException, TimeoutException {
- int partitionCount = fetchPartitionCount(timeoutMillis);
+
+ int partitionCount;
+ try {
+ partitionCount = fetchPartitionCount(timeoutMillis);
Review Comment:
IIUC, this part is processing a single stream. If we want to handle the case of multiple topics with one topic deleted, we should handle it on the caller side instead.
--
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
Re: [PR] Fix: Handle kafka topic not found exception in fetchPartitionCount fu… [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #17088:
URL: https://github.com/apache/pinot/pull/17088#issuecomment-3544304165
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/17088?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `23.07692%` with `10 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 33.58%. Comparing base ([`51632a5`](https://app.codecov.io/gh/apache/pinot/commit/51632a51ff515b4cd84b226fca49a7b81b3884db?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`9999d5e`](https://app.codecov.io/gh/apache/pinot/commit/9999d5eab8dabe68d00126ec61aa893bf590cb24?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 98 commits behind head on master.
:white_check_mark: All tests successful. No failed tests found.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/17088?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...pache/pinot/spi/stream/StreamMetadataProvider.java](https://app.codecov.io/gh/apache/pinot/pull/17088?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fstream%2FStreamMetadataProvider.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvc3RyZWFtL1N0cmVhbU1ldGFkYXRhUHJvdmlkZXIuamF2YQ==) | 23.07% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/17088?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (51632a5) and HEAD (9999d5e). Click for more details.
>
> <details><summary>HEAD has 28 uploads less than BASE</summary>
>
>| Flag | BASE (51632a5) | HEAD (9999d5e) |
>|------|------|------|
>|unittests1|1|0|
>|unittests|3|1|
>|java-11|4|0|
>|temurin|8|1|
>|unittests2|2|1|
>|java-21|4|1|
>|integration|5|0|
>|integration2|2|0|
>|integration1|2|0|
>|custom-integration1|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #17088 +/- ##
=============================================
- Coverage 63.17% 33.58% -29.60%
+ Complexity 1421 720 -701
=============================================
Files 3104 3104
Lines 183247 183366 +119
Branches 28088 28098 +10
=============================================
- Hits 115774 61584 -54190
- Misses 58502 116698 +58196
+ Partials 8971 5084 -3887
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.58% <23.07%> (+0.09%)` | :arrow_up: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.58% <23.07%> (-29.60%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.58% <23.07%> (-29.60%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/17088/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.58% <23.07%> (+0.09%)` | :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.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/17088?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</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@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
Re: [PR] Fix: Handle kafka topic not found exception in fetchPartitionCount fu… [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17088:
URL: https://github.com/apache/pinot/pull/17088#discussion_r2467640453
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java:
##########
@@ -86,7 +87,24 @@ StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria offsetCriteri
default List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
throws IOException, TimeoutException {
- int partitionCount = fetchPartitionCount(timeoutMillis);
+
+ int partitionCount;
+ try {
+ partitionCount = fetchPartitionCount(timeoutMillis);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to fetch partition count for stream config: {}. Skipping stream and using"
+ + "existing partitions only. Error: {}", streamConfig.getTopicName(), e.getMessage(), e);
Review Comment:
The log message is split across two lines with a string concatenation using `+`. This should be written as a single continuous string literal for better readability and to avoid the concatenation operator.
```suggestion
LOGGER.warn("Failed to fetch partition count for stream config: {}. Skipping stream and using existing partitions only. Error: {}", streamConfig.getTopicName(), e.getMessage(), e);
```
--
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
Re: [PR] Fix: Handle kafka topic not found exception in fetchPartitionCount fu… [pinot]
Posted by "Abhishek01911 (via GitHub)" <gi...@apache.org>.
Abhishek01911 commented on code in PR #17088:
URL: https://github.com/apache/pinot/pull/17088#discussion_r2471360784
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java:
##########
@@ -86,7 +87,24 @@ StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria offsetCriteri
default List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
throws IOException, TimeoutException {
- int partitionCount = fetchPartitionCount(timeoutMillis);
+
+ int partitionCount;
+ try {
+ partitionCount = fetchPartitionCount(timeoutMillis);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to fetch partition count for stream config: {}. Skipping stream and using"
+ + "existing partitions only. Error: {}", streamConfig.getTopicName(), e.getMessage(), e);
+ // Return only the existing partition groups if we can't fetch partition count
+ List<PartitionGroupMetadata> existingPartitionGroupMetadataList =
+ new ArrayList<>(partitionGroupConsumptionStatuses.size());
+ for (PartitionGroupConsumptionStatus currentPartitionGroupConsumptionStatus : partitionGroupConsumptionStatuses) {
+ existingPartitionGroupMetadataList.add(
+ new PartitionGroupMetadata(currentPartitionGroupConsumptionStatus.getStreamPartitionGroupId(),
+ currentPartitionGroupConsumptionStatus.getEndOffset()));
Review Comment:
added a comment for it. endOffset is only set as start offset if the last consuming segment commited.
It will be equal to previous commit segment end offset
--
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