You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "lnbest0707-uber (via GitHub)" <gi...@apache.org> on 2025/08/04 06:00:28 UTC
[PR] Introduce ephemeral topic type into multi-topic ingestion [pinot]
lnbest0707-uber opened a new pull request, #16494:
URL: https://github.com/apache/pinot/pull/16494
`real-time` `ingestion` `feature`
Part 2 of https://github.com/apache/pinot/pull/15782
Issue https://github.com/apache/pinot/issues/14815
Design doc https://docs.google.com/document/d/1NKPeNh6V2ctaQ4T_X3OKJ6Gcy5TRanLiU1uIDT8_9UA/edit?usp=sharing
Introduce the new ephemeral type of topic in multi-topic ingestion. The main difference would be the segment name. **For such type of topic, the segment name would be `tableName__topicName__partitionGroupId__sequenceNumber__creationTime` instead of `tableName__partitionGroupId__sequenceNumber__creationTime`.**
This change is supposed to be no-op at the moment because such naming would not be applied to any ingestions. Afterwards, this would be used to:
- [Short-term]Support skip interval backfill for offset auto reset during real-time ingestion lag (configurable).
- [Mid/Long-term]Replace the existing multi-topic ingestion naming format. So that it would be easy to remove topics from multi-topic table.
Other changes:
- Ensure each stream could use their own streamConfig instead of sharing the configs from the first streamConfig. Removing the streamConfigs.get(0) usage.
- Does not allow duplicate stream (same stream type with same topic name)
- [Tentative] Does not allow deleting topics. Deleting topics is not banned currently, but it is not a safe operation as segment metadata mapping would be messed.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2294858346
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java:
##########
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
+ topicName;
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
int index = i;
+ int finalPermanentTopicIndex = permanentTopicIndex;
+ // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+ // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
_partitionGroupConsumptionStatusList.stream()
- .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
- partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+ .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
+ ? _streamConfigs.get(index).getTopicName().equals(partitionGroupConsumptionStatus.getTopicName())
Review Comment:
Code refactored, pls check again, thanks
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2294832628
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2776,54 +2842,4 @@ public List<String> getCommittingSegments(String realtimeTableName) {
private List<String> getCommittingSegments(String realtimeTableName, Collection<String> segmentsToCheck) {
return getCommittingSegments(realtimeTableName, segmentsToCheck, _helixResourceManager::getSegmentZKMetadata);
}
-
- public static List<String> getCommittingSegments(String realtimeTableName,
Review Comment:
The IDE moved the static methods before non-static ones. We do not have strict enforcement in Pinot styles. If preferred not to be change, I can move them back.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3220960979
> > One possible encoding scheme is to encode topic name, backfill id and partition id in the max allowed int 2,147,483,647 limit. The first 4 digits are for topic, next 2 digit for backfill id and the next 4 digits for partition id. I think it should accommodate for nearly possible practical needs.
>
> I can think about 2 encoding methods but neither could fulfill the requirement
>
> * Use the sequence number as backfill or topic id -> It is hard to remove the topic.
The backfill topic id is encoded in the partitioned id today. E.g., 11010005 means the 5th partition of the 11th topic (in the topic map) and it is a backfill topic (so 01). This is the way we can detect a backfill topic.
> * Use the hardcode to map string to int id -> It is hard to avoid and deal with the collision.
Today in multi-topic, we use the index of the topic in the streamConfig map to map a topic to a int. The above proposal uses the same idea. Any issue with continuing today's practice.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3222995768
Added an alternative solution in https://github.com/apache/pinot/pull/16692
Will let the community to vote for the solution.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2294823224
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1833,29 +1896,31 @@ private StreamPartitionMsgOffset selectStartOffset(OffsetCriteria offsetCriteria
}
private LLCSegmentName getNextLLCSegmentName(LLCSegmentName lastLLCSegmentName, long creationTimeMs) {
- return new LLCSegmentName(lastLLCSegmentName.getTableName(), lastLLCSegmentName.getPartitionGroupId(),
- lastLLCSegmentName.getSequenceNumber() + 1, creationTimeMs);
+ return new LLCSegmentName(lastLLCSegmentName.getTableName(), lastLLCSegmentName.getTopicName(),
+ lastLLCSegmentName.getPartitionGroupId(), lastLLCSegmentName.getSequenceNumber() + 1, creationTimeMs);
}
/**
* Sets up a new partition group.
* <p>Persists the ZK metadata for the first CONSUMING segment, and returns the segment name.
*/
- private String setupNewPartitionGroup(TableConfig tableConfig, StreamConfig streamConfig,
+ private String setupNewPartitionGroup(TableConfig tableConfig, List<StreamConfig> streamConfigs,
Review Comment:
If you check the usage of `setupNewPartitionGroup`, they were all doing `setupNewPartitionGroup(xx, streamConfigs.get(0), xxx)`.
The function mainly relies on `partitionGroupMetadata` to setup new partition. And the `streamConfig` is only used for fetching the flush threshold parameters. And it is using the 1st config's instead of the corresponding's.
With this change, we would pass in the entire table config ('s streamConfigMaps). And the followup usages can pick the correct `streamConfig` by cross checking the `partitionGroupMetadata` and use their own flush threshold values.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3156141745
@Jackie-Jiang @noob-se7en @KKcorps please help review, thanks.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber closed pull request #16494: [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion
URL: https://github.com/apache/pinot/pull/16494
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3230184552
Abandon this one as the other option in https://github.com/apache/pinot/pull/16692 is preferred.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3192407461
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/16494?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 `31.63265%` with `67 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.37%. Comparing base ([`3d44c91`](https://app.codecov.io/gh/apache/pinot/commit/3d44c91e3c034c54daaeab8d0bceea20e7000ca6?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`f379825`](https://app.codecov.io/gh/apache/pinot/commit/f3798254fdeed91df4ab6dd7f6a9c09fec29f4fa?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/16494?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../org/apache/pinot/common/utils/LLCSegmentName.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcommon%2Futils%2FLLCSegmentName.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vdXRpbHMvTExDU2VnbWVudE5hbWUuamF2YQ==) | 55.00% | [15 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...inot/spi/stream/PartitionGroupMetadataFetcher.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fstream%2FPartitionGroupMetadataFetcher.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvc3RyZWFtL1BhcnRpdGlvbkdyb3VwTWV0YWRhdGFGZXRjaGVyLmphdmE=) | 0.00% | [14 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...g/apache/pinot/spi/utils/IngestionConfigUtils.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Futils%2FIngestionConfigUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvdXRpbHMvSW5nZXN0aW9uQ29uZmlnVXRpbHMuamF2YQ==) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/pinot/spi/stream/PartitionGroupMetadata.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fstream%2FPartitionGroupMetadata.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvc3RyZWFtL1BhcnRpdGlvbkdyb3VwTWV0YWRhdGEuamF2YQ==) | 0.00% | [8 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ot/spi/stream/PartitionGroupConsumptionStatus.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fstream%2FPartitionGroupConsumptionStatus.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvc3RyZWFtL1BhcnRpdGlvbkdyb3VwQ29uc3VtcHRpb25TdGF0dXMuamF2YQ==) | 28.57% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...a/manager/realtime/RealtimeSegmentDataManager.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fdata%2Fmanager%2Frealtime%2FRealtimeSegmentDataManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvUmVhbHRpbWVTZWdtZW50RGF0YU1hbmFnZXIuamF2YQ==) | 33.33% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...he/pinot/segment/local/utils/TableConfigUtils.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Futils%2FTableConfigUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC91dGlscy9UYWJsZUNvbmZpZ1V0aWxzLmphdmE=) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...java/org/apache/pinot/spi/stream/StreamConfig.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fstream%2FStreamConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvc3RyZWFtL1N0cmVhbUNvbmZpZy5qYXZh) | 57.14% | [2 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../config/table/ingestion/StreamIngestionConfig.java](https://app.codecov.io/gh/apache/pinot/pull/16494?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fconfig%2Ftable%2Fingestion%2FStreamIngestionConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvY29uZmlnL3RhYmxlL2luZ2VzdGlvbi9TdHJlYW1Jbmdlc3Rpb25Db25maWcuamF2YQ==) | 33.33% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16494?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 (3d44c91) and HEAD (f379825). Click for more details.
>
> <details><summary>HEAD has 32 uploads less than BASE</summary>
>
>| Flag | BASE (3d44c91) | HEAD (f379825) |
>|------|------|------|
>|java-21|4|1|
>|unittests1|2|1|
>|unittests|4|1|
>|temurin|9|1|
>|unittests2|2|0|
>|java-11|5|0|
>|integration|5|0|
>|integration1|2|0|
>|custom-integration1|2|0|
>|integration2|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #16494 +/- ##
============================================
- Coverage 63.37% 56.37% -7.00%
+ Complexity 1379 699 -680
============================================
Files 3020 2364 -656
Lines 175601 130752 -44849
Branches 26918 20819 -6099
============================================
- Hits 111280 73716 -37564
+ Misses 55828 50951 -4877
+ Partials 8493 6085 -2408
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/16494/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/16494/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/16494/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/16494/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/16494/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/16494/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/16494/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.37% <31.63%> (-6.92%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/16494/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.37% <31.63%> (-7.00%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/16494/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.37% <31.63%> (-6.99%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/16494/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.37% <31.63%> (-0.10%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/16494/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 Sentry](https://app.codecov.io/gh/apache/pinot/pull/16494?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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3218933366
> One possible encoding scheme is to encode topic name, backfill id and partition id in the max allowed int 2,147,483,647 limit. The first 4 digits are for topic, next 2 digit for backfill id and the next 4 digits for partition id. I think it should accommodate for nearly possible practical needs.
I can think about 2 encoding methods but neither could fulfill the requirement
- Use the sequence number as backfill or topic id -> It is hard to remove the topic.
- Use the hardcode to map string to int id -> It is hard to avoid and deal with the collision.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3207078142
@lnbest0707-uber can you please add problem and solution description just specific to this PR in the PR description.
If I understand correctly, this PR adds logic to consume from new topic type and this topic will have data which was skipped before by controller.
1. If above is correct can you tell how is this topic created in real time in upstream.
2. Also why we have to make changes to LLCSegmentName?
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2283182675
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,62 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = "";
Review Comment:
Can we use `null` to represent unavailable instead of using empty string? It is usually easier to maintain with `null`
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,62 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = "";
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, "", partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
+ String tableName, String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
+ Preconditions.checkArgument(topicName == null || !topicName.contains(SEPARATOR),
+ "Illegal topic name: %s", tableName);
_tableName = tableName;
+ _topicName = topicName;
_partitionGroupId = partitionGroupId;
_sequenceNumber = sequenceNumber;
// ISO8601 date: 20160120T1234Z
_creationTime = DATE_FORMATTER.print(msSinceEpoch);
- _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+ if ("".equals(topicName)) {
+ _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+ } else {
+ _segmentName =
Review Comment:
This breaks when `topicName` is `null`. This is an example why we don't want to mix handling of `null` and empty string
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -94,17 +129,34 @@ public static boolean isLowLevelConsumerSegmentName(String segmentName) {
* Returns the sequence number of the given segment name.
*/
public static int getSequenceNumber(String segmentName) {
- return Integer.parseInt(StringUtils.splitByWholeSeparator(segmentName, SEPARATOR)[2]);
+ String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
+ if (parts.length == 4) {
+ return Integer.parseInt(parts[2]);
+ } else {
+ return Integer.parseInt(parts[3]);
+ }
}
public String getTableName() {
return _tableName;
}
+ public String getTopicName() {
Review Comment:
Annotate the return as `@Nullable`
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,62 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = "";
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, "", partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
+ String tableName, String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
Review Comment:
Annotate `topicName` with `@Nullable`
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -94,17 +129,34 @@ public static boolean isLowLevelConsumerSegmentName(String segmentName) {
* Returns the sequence number of the given segment name.
*/
public static int getSequenceNumber(String segmentName) {
- return Integer.parseInt(StringUtils.splitByWholeSeparator(segmentName, SEPARATOR)[2]);
+ String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
+ if (parts.length == 4) {
+ return Integer.parseInt(parts[2]);
+ } else {
+ return Integer.parseInt(parts[3]);
+ }
}
public String getTableName() {
return _tableName;
}
+ public String getTopicName() {
+ return _topicName;
+ }
+
public int getPartitionGroupId() {
return _partitionGroupId;
}
+ public String getPartitionGroupInfo() {
Review Comment:
This can cause confusion. Can you move the topic name check to caller side?
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2292310357
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,63 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ @Nullable
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = null;
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, null, partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
+ String tableName, @Nullable String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
+ Preconditions.checkArgument(topicName == null || !topicName.contains(SEPARATOR),
Review Comment:
Consider also checking if it is not empty string
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,63 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ @Nullable
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = null;
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, null, partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
+ String tableName, @Nullable String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
+ Preconditions.checkArgument(topicName == null || !topicName.contains(SEPARATOR),
+ "Illegal topic name: %s", tableName);
Review Comment:
```suggestion
"Illegal topic name: %s", topicName);
```
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,63 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ @Nullable
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = null;
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, null, partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
+ String tableName, @Nullable String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
+ Preconditions.checkArgument(topicName == null || !topicName.contains(SEPARATOR),
+ "Illegal topic name: %s", tableName);
_tableName = tableName;
+ _topicName = topicName;
_partitionGroupId = partitionGroupId;
_sequenceNumber = sequenceNumber;
// ISO8601 date: 20160120T1234Z
_creationTime = DATE_FORMATTER.print(msSinceEpoch);
- _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+ if (topicName == null) {
+ _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+ } else {
+ _segmentName =
+ tableName + SEPARATOR + topicName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR
+ + _creationTime;
+ }
+ }
+
+ private LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, String creationTime,
Review Comment:
Remove this constructor
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,63 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ @Nullable
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = null;
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, null, partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
Review Comment:
(minor, format) The changes doesn't follow [Pinot Style](https://docs.pinot.apache.org/developers/developers-and-contributors/code-setup#set-up-ide). Can you set it up and reformat the changes
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -76,13 +118,7 @@ public static LLCSegmentName of(String segmentName) {
* Returns whether the given segment name represents an LLC segment.
*/
public static boolean isLLCSegment(String segmentName) {
- int numSeparators = 0;
- int index = 0;
- while ((index = segmentName.indexOf(SEPARATOR, index)) != -1) {
- numSeparators++;
- index += 2; // SEPARATOR.length()
- }
- return numSeparators == 3;
+ return of(segmentName) != null;
Review Comment:
Let's only change the last check as `numSeparators == 3 || numSeparators == 4` as it is more efficient
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2776,54 +2842,4 @@ public List<String> getCommittingSegments(String realtimeTableName) {
private List<String> getCommittingSegments(String realtimeTableName, Collection<String> segmentsToCheck) {
return getCommittingSegments(realtimeTableName, segmentsToCheck, _helixResourceManager::getSegmentZKMetadata);
}
-
- public static List<String> getCommittingSegments(String realtimeTableName,
Review Comment:
Several methods/comments are moved. What's the reason? Seems these methods fit more in their original position
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -127,6 +181,11 @@ public String getSegmentName() {
public int compareTo(LLCSegmentName other) {
Preconditions.checkArgument(_tableName.equals(other._tableName),
"Cannot compare segment names from different table: %s, %s", _segmentName, other.getSegmentName());
+ String thisTopicName = _topicName == null ? "" : _topicName;
+ String otherTopicName = other._topicName == null ? "" : other._topicName;
+ if (!thisTopicName.equals(otherTopicName)) {
+ return StringUtils.compare(_topicName, other._topicName);
+ }
Review Comment:
Do you want to compare topic name here? What is the side effect of comparing it?
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -65,6 +103,10 @@ public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber
*/
@Nullable
public static LLCSegmentName of(String segmentName) {
+ String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
+ if (parts.length < 4 || parts.length > 5) {
+ return null;
+ }
Review Comment:
Let's not modify this. This is adding overhead
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java:
##########
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
+ topicName;
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
int index = i;
+ int finalPermanentTopicIndex = permanentTopicIndex;
+ // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+ // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
_partitionGroupConsumptionStatusList.stream()
- .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
- partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+ .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
Review Comment:
Can we move this check to `PinotLLCRealtimeSegmentManager.setupNewPartitionGroup()`? Currently it is deeply nested, which makes it very hard to follow (I missed it several times, and in the end find this rewrite)
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1833,29 +1896,31 @@ private StreamPartitionMsgOffset selectStartOffset(OffsetCriteria offsetCriteria
}
private LLCSegmentName getNextLLCSegmentName(LLCSegmentName lastLLCSegmentName, long creationTimeMs) {
- return new LLCSegmentName(lastLLCSegmentName.getTableName(), lastLLCSegmentName.getPartitionGroupId(),
- lastLLCSegmentName.getSequenceNumber() + 1, creationTimeMs);
+ return new LLCSegmentName(lastLLCSegmentName.getTableName(), lastLLCSegmentName.getTopicName(),
+ lastLLCSegmentName.getPartitionGroupId(), lastLLCSegmentName.getSequenceNumber() + 1, creationTimeMs);
}
/**
* Sets up a new partition group.
* <p>Persists the ZK metadata for the first CONSUMING segment, and returns the segment name.
*/
- private String setupNewPartitionGroup(TableConfig tableConfig, StreamConfig streamConfig,
+ private String setupNewPartitionGroup(TableConfig tableConfig, List<StreamConfig> streamConfigs,
Review Comment:
I don't follow this change. What does it mean to setup a new partition group on multiple `StreamConfig`? Which `StreamConfig` should we reference when setting it up?
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java:
##########
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
+ topicName;
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
int index = i;
+ int finalPermanentTopicIndex = permanentTopicIndex;
+ // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+ // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
_partitionGroupConsumptionStatusList.stream()
- .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
- partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+ .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
+ ? _streamConfigs.get(index).getTopicName().equals(partitionGroupConsumptionStatus.getTopicName())
+ : IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
+ partitionGroupConsumptionStatus.getPartitionGroupId()) == finalPermanentTopicIndex)
.collect(Collectors.toList());
try (StreamMetadataProvider streamMetadataProvider = streamConsumerFactory.createStreamMetadataProvider(
StreamConsumerFactory.getUniqueClientId(clientId))) {
+ // Similarly, for ephemeral backfill topics, we create the partition group metadata with the topic name.
_newPartitionGroupMetadataList.addAll(
streamMetadataProvider.computePartitionGroupMetadata(clientId,
- streamConfig, topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
+ _streamConfigs.get(i), topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
_forceGetOffsetFromStream)
.stream()
.map(metadata -> new PartitionGroupMetadata(
- IngestionConfigUtils.getPinotPartitionIdFromStreamPartitionId(metadata.getPartitionGroupId(),
- index), metadata.getStartOffset()))
+ _streamConfigs.get(finalPermanentTopicIndex).isEphemeralBackfillTopic() ? _streamConfigs.get(index)
+ .getTopicName() : "",
Review Comment:
Is this a bug? Will it break if we use `""` as the topic name? Is there a test for this?
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java:
##########
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
+ topicName;
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
int index = i;
+ int finalPermanentTopicIndex = permanentTopicIndex;
+ // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+ // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
_partitionGroupConsumptionStatusList.stream()
- .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
- partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+ .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
+ ? _streamConfigs.get(index).getTopicName().equals(partitionGroupConsumptionStatus.getTopicName())
Review Comment:
Do we need this check?
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "chenboat (via GitHub)" <gi...@apache.org>.
chenboat commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3216035783
Given most of this PR is related to segment name format change, plus the fact that LLC segment format is used and checked in many Pinot classes. Is it possible to continue the current format but instead of encode the topic name in the current format? Put topic names in the segment name also can result in segment name length explosion and related metadata increase.
One possible encoding scheme is to encode topic name, backfill id and partition id in the max allowed int 2,147,483,647 limit.
The first 4 digits are for topic, next 2 digit for backfill id and the next 4 digits for partition id. I think it should accommodate for nearly possible practical needs.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2283698971
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -94,17 +129,34 @@ public static boolean isLowLevelConsumerSegmentName(String segmentName) {
* Returns the sequence number of the given segment name.
*/
public static int getSequenceNumber(String segmentName) {
- return Integer.parseInt(StringUtils.splitByWholeSeparator(segmentName, SEPARATOR)[2]);
+ String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
+ if (parts.length == 4) {
+ return Integer.parseInt(parts[2]);
+ } else {
+ return Integer.parseInt(parts[3]);
+ }
}
public String getTableName() {
return _tableName;
}
+ public String getTopicName() {
+ return _topicName;
+ }
+
public int getPartitionGroupId() {
return _partitionGroupId;
}
+ public String getPartitionGroupInfo() {
Review Comment:
Change the function name to make it more intuitive
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3207106754
> @lnbest0707-uber can you please add problem and solution description just specific to this PR in the PR description. If I understand correctly, this PR adds logic to consume from new topic type and this topic will have data which was skipped before by controller.
>
> 1. If above is correct can you tell how is this topic created in real time in upstream.
> 2. Also why we have to make changes to LLCSegmentName?
The PR description is exactly for this PR only. You can forget about the offset auto reset story. The PR is about multi-topic ingestion.
Previously, we uses the 10000 number to pad/map between stream partition Id to Pinot partition Id. (e.g. partition 1 of the second topic would have 10001 partition id in Pinot).
With this PR, we support a special type of topic in multi-topic ingestion. And the mapping between stream partition id and Pinot partition id would be `<topic_name>__<stream_partitionId>`.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2292310679
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfig.java:
##########
@@ -431,6 +435,10 @@ public long getOffsetAutoResetTimeSecThreshold() {
return _offsetAutoResetTimeSecThreshold;
}
+ public Boolean isEphemeralBackfillTopic() {
+ return Boolean.TRUE.equals(_ephemeralBackfillTopic);
Review Comment:
The return statement uses Boolean.TRUE.equals() which is unnecessarily complex. Since _ephemeralBackfillTopic is initialized to false when null, a simple null check and return would be clearer: return _ephemeralBackfillTopic != null && _ephemeralBackfillTopic;
```suggestion
return _ephemeralBackfillTopic != null && _ephemeralBackfillTopic;
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2352,16 +2417,17 @@ private Set<String> filterSegmentsToCommit(Set<String> allConsumingSegments,
return segmentsToCommit;
}
- // partitionGroupIdsToCommitStr != null
- Set<Integer> partitionsToCommit = Arrays.stream(partitionGroupIdsToCommitStr.split(","))
+ // partitionGroupInfosToCommitStr != null
+ Set<Integer> partitionsToCommit = Arrays.stream(partitionGroupInfosToCommitStr.split(","))
.map(String::trim)
.map(Integer::parseInt)
.collect(Collectors.toSet());
Set<String> targetSegments = allConsumingSegments.stream()
- .filter(segmentName -> partitionsToCommit.contains(new LLCSegmentName(segmentName).getPartitionGroupId()))
+ .filter(
+ segmentName -> partitionsToCommit.contains(new LLCSegmentName(segmentName).getPartitionGroupTopicAndId()))
Review Comment:
The filter is calling getPartitionGroupTopicAndId() but comparing against partitionsToCommit which contains Integer values parsed from comma-separated string. This will cause ClassCastException since String cannot be compared to Integer.
```suggestion
segmentName -> partitionsToCommit.contains(new LLCSegmentName(segmentName).getPartitionGroupId()))
```
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,63 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ @Nullable
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = null;
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, null, partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
+ String tableName, @Nullable String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
+ Preconditions.checkArgument(topicName == null || !topicName.contains(SEPARATOR),
+ "Illegal topic name: %s", tableName);
_tableName = tableName;
+ _topicName = topicName;
_partitionGroupId = partitionGroupId;
_sequenceNumber = sequenceNumber;
// ISO8601 date: 20160120T1234Z
_creationTime = DATE_FORMATTER.print(msSinceEpoch);
- _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+ if (topicName == null) {
+ _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+ } else {
+ _segmentName =
+ tableName + SEPARATOR + topicName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR
+ + _creationTime;
+ }
+ }
+
+ private LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, String creationTime,
+ String segmentName) {
+ _tableName = tableName;
+ _topicName = "";
Review Comment:
The private constructor sets _topicName to an empty string instead of null, which is inconsistent with the null check pattern used elsewhere in the class. This could cause issues with the getPartitionGroupTopicAndId() method logic.
```suggestion
_topicName = null;
```
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java:
##########
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
+ topicName;
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
int index = i;
+ int finalPermanentTopicIndex = permanentTopicIndex;
+ // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+ // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
_partitionGroupConsumptionStatusList.stream()
- .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
- partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+ .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
+ ? _streamConfigs.get(index).getTopicName().equals(partitionGroupConsumptionStatus.getTopicName())
+ : IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
+ partitionGroupConsumptionStatus.getPartitionGroupId()) == finalPermanentTopicIndex)
.collect(Collectors.toList());
try (StreamMetadataProvider streamMetadataProvider = streamConsumerFactory.createStreamMetadataProvider(
StreamConsumerFactory.getUniqueClientId(clientId))) {
+ // Similarly, for ephemeral backfill topics, we create the partition group metadata with the topic name.
_newPartitionGroupMetadataList.addAll(
streamMetadataProvider.computePartitionGroupMetadata(clientId,
- streamConfig, topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
+ _streamConfigs.get(i), topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
_forceGetOffsetFromStream)
.stream()
.map(metadata -> new PartitionGroupMetadata(
- IngestionConfigUtils.getPinotPartitionIdFromStreamPartitionId(metadata.getPartitionGroupId(),
- index), metadata.getStartOffset()))
+ _streamConfigs.get(finalPermanentTopicIndex).isEphemeralBackfillTopic() ? _streamConfigs.get(index)
+ .getTopicName() : "",
+ _streamConfigs.get(finalPermanentTopicIndex).isEphemeralBackfillTopic()
Review Comment:
The condition logic is incorrect. It's checking if the stream config at finalPermanentTopicIndex is ephemeral, but it should be checking if the stream config at index is ephemeral. This could lead to incorrect partition group metadata creation.
```suggestion
_streamConfigs.get(index).isEphemeralBackfillTopic() ? _streamConfigs.get(index)
.getTopicName() : "",
_streamConfigs.get(index).isEphemeralBackfillTopic()
```
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on PR #16494:
URL: https://github.com/apache/pinot/pull/16494#issuecomment-3221006399
> E.g., 11010005 means the 5th partition of the 11th topic (in the topic map) and it is a backfill topic (so 01). This is the way we can detect a backfill topic.
With this naming format, if we remove a topic, it might affect other topic's pinot partition naming format.
Let's say there's we have 11th topic A as backfill topic, 1101000x, 12th topic B also backfill topic, 1201000x. Then when you remove topicA, how should we naming the topic B. With the naming format, it would become 1101000x, which would be messed up.
> Today in multi-topic, we use the index of the topic in the streamConfig map to map a topic to a int. The above proposal uses the same idea. Any issue with continuing today's practice.
As above example, the issue is about **removing topics**. This index mapping makes topic partition ID rely on its **position**. And when other topics got removed, the position would change.
--
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] [Auto reset 2/3]Introduce ephemeral topic type into multi-topic ingestion [pinot]
Posted by "lnbest0707-uber (via GitHub)" <gi...@apache.org>.
lnbest0707-uber commented on code in PR #16494:
URL: https://github.com/apache/pinot/pull/16494#discussion_r2292380542
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -38,25 +38,63 @@ public class LLCSegmentName implements Comparable<LLCSegmentName> {
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
+ @Nullable
+ private final String _topicName;
public LLCSegmentName(String segmentName) {
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
- Preconditions.checkArgument(parts.length == 4, "Invalid LLC segment name: %s", segmentName);
+ // Validate the segment name format should have 4 or 5 parts:
+ // e.g. tableName__partitionGroupId__sequenceNumber__creationTime
+ // or tableName__topicName__partitionGroupId__sequenceNumber__creationTime
+ Preconditions.checkArgument(
+ parts.length >= 4 && parts.length <= 5, "Invalid LLC segment name: %s", segmentName);
_tableName = parts[0];
- _partitionGroupId = Integer.parseInt(parts[1]);
- _sequenceNumber = Integer.parseInt(parts[2]);
- _creationTime = parts[3];
+ if (parts.length == 4) {
+ _topicName = null;
+ _partitionGroupId = Integer.parseInt(parts[1]);
+ _sequenceNumber = Integer.parseInt(parts[2]);
+ _creationTime = parts[3];
+ } else {
+ _topicName = parts[1];
+ _partitionGroupId = Integer.parseInt(parts[2]);
+ _sequenceNumber = Integer.parseInt(parts[3]);
+ _creationTime = parts[4];
+ }
_segmentName = segmentName;
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
+ this(tableName, null, partitionGroupId, sequenceNumber, msSinceEpoch);
+ }
+
+ public LLCSegmentName(
+ String tableName, @Nullable String topicName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
+ Preconditions.checkArgument(topicName == null || !topicName.contains(SEPARATOR),
Review Comment:
good idea
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -76,13 +118,7 @@ public static LLCSegmentName of(String segmentName) {
* Returns whether the given segment name represents an LLC segment.
*/
public static boolean isLLCSegment(String segmentName) {
- int numSeparators = 0;
- int index = 0;
- while ((index = segmentName.indexOf(SEPARATOR, index)) != -1) {
- numSeparators++;
- index += 2; // SEPARATOR.length()
- }
- return numSeparators == 3;
+ return of(segmentName) != null;
Review Comment:
good point.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java:
##########
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
+ topicName;
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
int index = i;
+ int finalPermanentTopicIndex = permanentTopicIndex;
+ // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+ // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
_partitionGroupConsumptionStatusList.stream()
- .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
- partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+ .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
Review Comment:
Agree. This code is hard to follow. I've refactored it to completely split the permanent and ephemeral.
Also this class was with only simple functionality and without UT. I've also added extra UTs to test single/mutiple w/ and w/o ephemeral topics cases.
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java:
##########
@@ -127,6 +181,11 @@ public String getSegmentName() {
public int compareTo(LLCSegmentName other) {
Preconditions.checkArgument(_tableName.equals(other._tableName),
"Cannot compare segment names from different table: %s, %s", _segmentName, other.getSegmentName());
+ String thisTopicName = _topicName == null ? "" : _topicName;
+ String otherTopicName = other._topicName == null ? "" : other._topicName;
+ if (!thisTopicName.equals(otherTopicName)) {
+ return StringUtils.compare(_topicName, other._topicName);
+ }
Review Comment:
If we do not compare it. `table__topicA__1__xxTxxZ` might be same as `table__1__xxTxxZ`.
Do you mean if we should treat it as table name as above?
##########
pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java:
##########
@@ -107,21 +108,32 @@ private Boolean fetchMultipleStreams()
+ topicName;
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
int index = i;
+ int finalPermanentTopicIndex = permanentTopicIndex;
+ // For permanent topics, we use the index of the stream config to get the partition group consumption status.
+ // For ephemeral backfill topics, we use the topic name to filter the partition group consumption status.
List<PartitionGroupConsumptionStatus> topicPartitionGroupConsumptionStatusList =
_partitionGroupConsumptionStatusList.stream()
- .filter(partitionGroupConsumptionStatus -> IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
- partitionGroupConsumptionStatus.getPartitionGroupId()) == index)
+ .filter(partitionGroupConsumptionStatus -> _streamConfigs.get(index).isEphemeralBackfillTopic()
+ ? _streamConfigs.get(index).getTopicName().equals(partitionGroupConsumptionStatus.getTopicName())
+ : IngestionConfigUtils.getStreamConfigIndexFromPinotPartitionId(
+ partitionGroupConsumptionStatus.getPartitionGroupId()) == finalPermanentTopicIndex)
.collect(Collectors.toList());
try (StreamMetadataProvider streamMetadataProvider = streamConsumerFactory.createStreamMetadataProvider(
StreamConsumerFactory.getUniqueClientId(clientId))) {
+ // Similarly, for ephemeral backfill topics, we create the partition group metadata with the topic name.
_newPartitionGroupMetadataList.addAll(
streamMetadataProvider.computePartitionGroupMetadata(clientId,
- streamConfig, topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
+ _streamConfigs.get(i), topicPartitionGroupConsumptionStatusList, /*maxWaitTimeMs=*/15000,
_forceGetOffsetFromStream)
.stream()
.map(metadata -> new PartitionGroupMetadata(
- IngestionConfigUtils.getPinotPartitionIdFromStreamPartitionId(metadata.getPartitionGroupId(),
- index), metadata.getStartOffset()))
+ _streamConfigs.get(finalPermanentTopicIndex).isEphemeralBackfillTopic() ? _streamConfigs.get(index)
+ .getTopicName() : "",
Review Comment:
Standardized all related component's _topicName field. We would treat empty as null now.
--
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