Fix: Handle kafka topic not found exception in fetchPartitionCount fu…#17669
Open
Abhishek01911 wants to merge 16 commits intoapache:masterfrom
Open
Fix: Handle kafka topic not found exception in fetchPartitionCount fu…#17669Abhishek01911 wants to merge 16 commits intoapache:masterfrom
Abhishek01911 wants to merge 16 commits intoapache:masterfrom
Conversation
Jackie-Jiang
added
kafka
cbalci
reviewed
Feb 9, 2026
cbalci
reviewed
Feb 13, 2026
Contributor
cbalci
left a comment
cbalci
left a comment
There was a problem hiding this comment.
I think it would be best to guard this logic with a stream config like skip.missing.topics=true which is false by default.
cbalci
reviewed
Feb 20, 2026
cbalci
reviewed
Feb 24, 2026
| public static final String GROUP_ID = "hlc.group.id"; | ||
| public static final String PARTITION_MSG_OFFSET_FACTORY_CLASS = "partition.offset.factory.class.name"; | ||
| public static final String TOPIC_CONSUMPTION_RATE_LIMIT = "topic.consumption.rate.limit"; | ||
| public static final String SKIP_MISSING_TOPICS = "skip.missing.topics"; |
Contributor
There was a problem hiding this comment.
This isn't a generic property really. Can we create a new section for multitopic related configs and place it there with proper prefix and explanation?
Something like
MULTITOPIC_SKIP_MISSING_TOPICS = "multitopic.skip.missing.topics";
cbalci
reviewed
Feb 24, 2026
Comment on lines
+131
to
+132
| // Check if the topic exists before fetching partition metadata | ||
| // Only perform this check if topic existence validation is enabled and topics were fetched |
Contributor
There was a problem hiding this comment.
Since we renamed the config (and the variable), please update the comment accordingly.
cbalci
reviewed
Feb 24, 2026
| throws Exception { | ||
| int numStreams = _streamConfigs.size(); | ||
|
|
||
| // Fetch available topics once and reuse across all streams (for topic existence validation) |
Contributor
There was a problem hiding this comment.
Please clarify this comment
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #17669 +/- ##
============================================
- Coverage 63.22% 63.21% -0.02%
+ Complexity 1499 1454 -45
============================================
Files 3174 3176 +2
Lines 190319 191029 +710
Branches 29080 29207 +127
============================================
+ Hits 120338 120764 +426
- Misses 60643 60859 +216
- Partials 9338 9406 +68
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
Abhishek01911
force-pushed
the
exception_fix1
branch
from
f082bb6 to
ff8c650
Compare
Abhishek01911
force-pushed
the
exception_fix1
branch
from
ff8c650 to
c2ba4eb
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Resolves Issue #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.
This change only impacts multi topic tables only when property - skip.missing.topics is set as true.
Testing
Tested on multi Topic table
Validated multi topic table was able to ingest data even after kafka topic was deleted / invalid kafka topic was added