You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "KKcorps (via GitHub)" <gi...@apache.org> on 2026/10/26 06:07:21 UTC
[PR] Let streams report ended partitions and stop reporting them as consuming [pinot]
KKcorps opened a new pull request, #20040:
URL: https://github.com/apache/pinot/pull/20040
When a stream partition ends, like a Kinesis shard that is fully read after a split or merge, Pinot never creates another consuming segment for it, and alerts on `llcPartitionConsuming` keep firing for a partition that is done. Today the controller only sees that the partition is missing from the stream metadata, so it cannot tell an ended partition from a missing one.
This lets the stream say it. `StreamMetadataProvider` gets a `fetchEndOfLifePartitionGroupIds` method that defaults to none, and Kinesis reports shards that are closed and consumed to their end. When the stream confirms it, at commit or in `RealtimeSegmentValidationManager`, the controller sets `segment.realtime.endOfPartitionGroup` in the last segment's ZK metadata. Nothing is kept in memory, so a restart or leadership move needs nothing rebuilt.
Every replica reads that ZK metadata on CONSUMING to ONLINE, so when the flag is set it removes `llcPartitionConsuming` instead of leaving it at 0, whether it committed, caught up or downloaded. The controller also emits a `partitionEndOfLife` gauge of 1 for these partitions, for alerts that fill missing series with 0. Streams that never report an ended partition, like Kafka, see no change.
Tests: new cases in `KinesisStreamMetadataProviderTest`, `PinotLLCRealtimeSegmentManagerTest`, `RealtimeSegmentDataManagerTest` and `SegmentZKMetadataTest`, and each fails without its fix. The Prometheus export test, spotless, checkstyle and license pass.
--
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] Let streams report ended partitions and stop reporting them as consuming [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #20040:
URL: https://github.com/apache/pinot/pull/20040#discussion_r4393600976
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2196,6 +2294,36 @@ && new LLCSegmentName(segmentEntry.getKey()).getPartitionGroupId() == partitionI
}
}
+ if (!endOfLifeCandidates.isEmpty()) {
+ Set<Integer> endOfLifePartitionGroupIds = fetchEndOfLifePartitionGroupIds(streamConfigs, endOfLifeCandidates);
+ for (SegmentZKMetadata segmentZKMetadata : endOfLifeCandidates) {
+ String segmentName = segmentZKMetadata.getSegmentName();
+ if (endOfLifePartitionGroupIds.contains(new LLCSegmentName(segmentName).getPartitionGroupId())
+ && markEndOfPartitionGroup(realtimeTableName, segmentName)) {
+ segmentZKMetadata.setEndOfPartitionGroup(true);
+ if (onlineEndOfLifeCandidates.contains(segmentName)) {
+ sendPartitionEndOfLifeMessage(realtimeTableName, segmentName);
Review Comment:
[P2] Reconcile cleanup for already-marked partitions
The terminal mark is persisted before this best-effort notification, but subsequent validation runs exclude marked segments at line 2210. `MessagingServiceUtils.send()` catches exceptions and returns zero, and `sendPartitionEndOfLifeMessage()` only logs the result. If delivery fails once, or the controller stops between persistence and sending, servers already ONLINE retain `LLC_PARTITION_CONSUMING=0` with no later cleanup attempt, preserving the false alerts this change addresses. Reconcile notifications for marked ONLINE segments or retain pending cleanup until it succeeds. Add a test where the first send fails and a later validation run repairs the server metrics.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -949,10 +949,95 @@ private String createNewSegmentMetadata(TableConfig tableConfig,
LOGGER.info(
"Skipping creation of new segment metadata after segment: {} during commit. Reason: Partition ID: {} not "
+ "found in upstream metadata.", committingSegmentName, committingSegmentPartitionGroupId);
+ if (fetchEndOfLifePartitionGroupIds(streamConfigs, List.of(committingSegmentZKMetadata)).contains(
+ committingSegmentPartitionGroupId)) {
+ markEndOfPartitionGroup(realtimeTableName, committingSegmentName);
+ }
}
return newConsumingSegmentName;
}
+ /// Asks the stream which of the given completed segments are the last segment of a partition group that reached end
+ /// of life (see [StreamMetadataProvider#fetchEndOfLifePartitionGroupIds]). Returns their partition group ids.
+ @VisibleForTesting
+ Set<Integer> fetchEndOfLifePartitionGroupIds(List<StreamConfig> streamConfigs,
+ Collection<SegmentZKMetadata> completedSegmentsZKMetadata) {
+ int numStreams = streamConfigs.size();
+ Map<Integer, List<PartitionGroupConsumptionStatus>> streamConfigIndexToStatuses = new HashMap<>();
+ for (SegmentZKMetadata segmentZKMetadata : completedSegmentsZKMetadata) {
+ if (segmentZKMetadata.getEndOffset() == null) {
+ continue;
+ }
+ LLCSegmentName llcSegmentName = new LLCSegmentName(segmentZKMetadata.getSegmentName());
+ int partitionGroupId = llcSegmentName.getPartitionGroupId();
+ // Same partition id translation as getPartitionGroupConsumptionStatusList()
+ int streamConfigIndex =
+ numStreams == 1 ? 0 : IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(partitionGroupId);
+ if (streamConfigIndex >= numStreams) {
+ continue;
+ }
+ int streamPartitionId = numStreams == 1 ? partitionGroupId
+ : IngestionConfigUtils.getStreamPartitionIdFromPinotPartitionId(partitionGroupId);
+ StreamPartitionMsgOffsetFactory offsetFactory =
+ StreamConsumerFactoryProvider.create(streamConfigs.get(streamConfigIndex)).createStreamMsgOffsetFactory();
+ streamConfigIndexToStatuses.computeIfAbsent(streamConfigIndex, k -> new ArrayList<>()).add(
+ new PartitionGroupConsumptionStatus(partitionGroupId, streamPartitionId, llcSegmentName.getSequenceNumber(),
+ offsetFactory.create(segmentZKMetadata.getStartOffset()),
+ offsetFactory.create(segmentZKMetadata.getEndOffset()), segmentZKMetadata.getStatus().toString()));
+ }
+ Set<Integer> endOfLifePartitionGroupIds = new HashSet<>();
+ for (Map.Entry<Integer, List<PartitionGroupConsumptionStatus>> entry : streamConfigIndexToStatuses.entrySet()) {
+ StreamConfig streamConfig = streamConfigs.get(entry.getKey());
+ try (StreamMetadataProvider metadataProvider = StreamConsumerFactoryProvider.create(streamConfig)
+ .createStreamMetadataProvider(getTableTopicUniqueClientId(streamConfig))) {
+ endOfLifePartitionGroupIds.addAll(
+ metadataProvider.fetchEndOfLifePartitionGroupIds(streamConfig, entry.getValue(),
+ (int) STREAM_FETCH_TIMEOUT_MS));
+ } catch (Exception e) {
+ LOGGER.warn("Failed to fetch end of life partition groups of topic: {} for table: {}",
+ streamConfig.getTopicName(), streamConfig.getTableNameWithType(), e);
+ }
+ }
+ return endOfLifePartitionGroupIds;
+ }
+
+ /// Marks the last segment of a partition group that reached end of life in its ZK metadata, so that servers stop
+ /// reporting the partition as consuming, and sets the PARTITION_END_OF_LIFE gauge. Returns whether it is marked.
+ private boolean markEndOfPartitionGroup(String realtimeTableName, String segmentName) {
+ try {
+ Stat stat = new Stat();
+ SegmentZKMetadata segmentZKMetadata = getSegmentZKMetadata(realtimeTableName, segmentName, stat);
+ if (!segmentZKMetadata.isEndOfPartitionGroup()) {
+ segmentZKMetadata.setEndOfPartitionGroup(true);
+ persistSegmentZKMetadata(realtimeTableName, segmentZKMetadata, stat.getVersion());
Review Comment:
[P2] Preserve the timestamp used by pauseless commit recovery
This can rewrite a segment while it remains COMMITTING, advancing the ZK modification time that `findNewSegmentsOfEarlierAttempts()` uses as the commit-start bound. For example, an attempt creates successor metadata at 12:00 and fails its IdealState update; a retry at 12:02 observes the now-ended shard, persists this flag, and fails before cleanup. The next retry searches from 12:01 and misses the original successor. Validation subsequently selects that orphan as the latest segment and can add it as CONSUMING despite the partition having ended, also emitting a false stream-data-loss count. Preserve a stable commit-start timestamp or make successor discovery independent of this mutable modification time. A regression test should cover marking terminal state between failed pauseless attempts across minute boundaries.
--
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] Let streams report ended partitions and stop reporting them as consuming [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20040:
URL: https://github.com/apache/pinot/pull/20040#issuecomment-6410145627
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/20040?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 `55.55556%` with `8 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 59.09%. Comparing base ([`bff525d`](https://app.codecov.io/gh/apache/pinot/commit/bff525d98ea7b1b2120777af4ad5c48a1089677f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`1c0ef4d`](https://app.codecov.io/gh/apache/pinot/commit/1c0ef4dc7a7effb3a2c7051b71f991a016da632f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/20040?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...g/apache/pinot/common/metrics/AbstractMetrics.java](https://app.codecov.io/gh/apache/pinot/pull/20040?src=pr&el=tree&filepath=pinot-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcommon%2Fmetrics%2FAbstractMetrics.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9BYnN0cmFjdE1ldHJpY3MuamF2YQ==) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/20040?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/pinot/spi/stream/StreamMetadataProvider.java](https://app.codecov.io/gh/apache/pinot/pull/20040?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==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/20040?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 (bff525d) and HEAD (1c0ef4d). Click for more details.
>
> <details><summary>HEAD has 14 uploads less than BASE</summary>
>
>| Flag | BASE (bff525d) | HEAD (1c0ef4d) |
>|------|------|------|
>|lane-a|1|0|
>|integration|2|0|
>|java-25|4|1|
>|integration1|2|0|
>|temurin|4|1|
>|lane-b|1|0|
>|unittests2|1|0|
>|unittests|2|1|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20040 +/- ##
=============================================
- Coverage 69.93% 59.09% -10.85%
+ Complexity 1519 1 -1518
=============================================
Files 3557 2742 -815
Lines 236305 173338 -62967
Branches 37943 28610 -9333
=============================================
- Hits 165256 102430 -62826
- Misses 58344 62255 +3911
+ Partials 12705 8653 -4052
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/20040/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/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `59.09% <55.55%> (-10.85%)` | :arrow_down: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `59.09% <55.55%> (-10.85%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `59.09% <55.55%> (-10.84%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `59.09% <55.55%> (-0.02%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/20040/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
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 Harness](https://app.codecov.io/gh/apache/pinot/pull/20040?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] Let streams report ended partitions and stop reporting them as consuming [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20040:
URL: https://github.com/apache/pinot/pull/20040#discussion_r4377006956
##########
pinot-common/src/main/java/org/apache/pinot/common/metrics/AbstractMetrics.java:
##########
@@ -685,6 +686,17 @@ public void removePartitionGauge(final String tableName, final int partitionId,
removeGauge(fullGaugeName);
}
+ /// Removes all partition gauges of the given gauge for the table.
+ public void removePartitionGauges(final String tableName, final G gauge) {
+ String prefix = composeTableGaugeName(tableName, gauge) + ".";
Review Comment:
Nit: when table-level metrics are disabled, `composeTableGaugeName` collapses the name to the `allTables` bucket, so this would clear the gauge for every table, not just this one. That matches the existing `removeTableGauge`/`removePartitionGauge` behavior in that mode, so it's consistent rather than a regression — just flagging it.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -949,10 +949,84 @@ private String createNewSegmentMetadata(TableConfig tableConfig,
LOGGER.info(
"Skipping creation of new segment metadata after segment: {} during commit. Reason: Partition ID: {} not "
+ "found in upstream metadata.", committingSegmentName, committingSegmentPartitionGroupId);
+ if (fetchEndOfLifePartitionGroupIds(streamConfigs, List.of(committingSegmentZKMetadata)).contains(
Review Comment:
Minor: this opens a fresh stream metadata provider and makes a synchronous Kinesis call inside the commit path. It only fires on the terminal commit of a partition that's already gone from upstream, so it's rare, but worth being aware it adds latency to that one commit. No change needed.
--
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