Jackie-Jiang commented on code in PR #19170:
URL: https://github.com/apache/pinot/pull/19170#discussion_r3832921034
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
Review Comment:
Update the javadoc here
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1386,32 +1371,31 @@ public void reduceSegmentSizeAndReset(LLCSegmentName
llcSegmentName, int prevNum
///
/// @param realtimeTableName Realtime table name
/// @return Map from partition group id to the latest LLC realtime segment
ZK metadata
- private Map<Integer, SegmentZKMetadata> getLatestSegmentZKMetadataMap(String
realtimeTableName) {
- List<String> segments = getLLCSegments(realtimeTableName);
-
- Map<Integer, LLCSegmentName> latestLLCSegmentNameMap = new HashMap<>();
- for (String segmentName : segments) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
- latestLLCSegmentNameMap.compute(llcSegmentName.getPartitionGroupId(),
(partitionId, latestLLCSegmentName) -> {
- if (latestLLCSegmentName == null) {
- return llcSegmentName;
- } else {
- if (llcSegmentName.getSequenceNumber() >
latestLLCSegmentName.getSequenceNumber()) {
- return llcSegmentName;
- } else {
- return latestLLCSegmentName;
- }
- }
- });
+ /// Returns the latest (highest sequence number) LLC segment per partition
group, parsed from the given segment
+ /// names. Non-LLC segment names (e.g. uploaded upsert segments) are ignored.
+ private static Map<Integer, LLCSegmentName>
getLatestLLCSegmentPerPartition(Collection<String> segmentNames) {
+ Map<Integer, LLCSegmentName> partitionGroupIdToLatestSegment = new
HashMap<>();
+ for (String segmentName : segmentNames) {
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName == null) {
+ continue;
+ }
+
partitionGroupIdToLatestSegment.merge(llcSegmentName.getPartitionGroupId(),
llcSegmentName,
+ (existing, candidate) -> candidate.getSequenceNumber() >
existing.getSequenceNumber() ? candidate
+ : existing);
}
+ return partitionGroupIdToLatestSegment;
+ }
+ private Map<Integer, SegmentZKMetadata> getLatestSegmentZKMetadataMap(String
realtimeTableName) {
+ Map<Integer, LLCSegmentName> latestLLCSegmentNameMap =
+ getLatestLLCSegmentPerPartition(getLLCSegments(realtimeTableName));
Map<Integer, SegmentZKMetadata> latestSegmentZKMetadataMap = new
HashMap<>();
Review Comment:
(minor) We can use a pre-sized map
```suggestion
Map<Integer, SegmentZKMetadata> latestSegmentZKMetadataMap =
Maps.newHashMapWithExpectedSize(latestLLCSegmentNameMap.size());
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1386,32 +1371,31 @@ public void reduceSegmentSizeAndReset(LLCSegmentName
llcSegmentName, int prevNum
///
/// @param realtimeTableName Realtime table name
/// @return Map from partition group id to the latest LLC realtime segment
ZK metadata
- private Map<Integer, SegmentZKMetadata> getLatestSegmentZKMetadataMap(String
realtimeTableName) {
- List<String> segments = getLLCSegments(realtimeTableName);
-
- Map<Integer, LLCSegmentName> latestLLCSegmentNameMap = new HashMap<>();
- for (String segmentName : segments) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
- latestLLCSegmentNameMap.compute(llcSegmentName.getPartitionGroupId(),
(partitionId, latestLLCSegmentName) -> {
- if (latestLLCSegmentName == null) {
- return llcSegmentName;
- } else {
- if (llcSegmentName.getSequenceNumber() >
latestLLCSegmentName.getSequenceNumber()) {
- return llcSegmentName;
- } else {
- return latestLLCSegmentName;
- }
- }
- });
+ /// Returns the latest (highest sequence number) LLC segment per partition
group, parsed from the given segment
+ /// names. Non-LLC segment names (e.g. uploaded upsert segments) are ignored.
+ private static Map<Integer, LLCSegmentName>
getLatestLLCSegmentPerPartition(Collection<String> segmentNames) {
+ Map<Integer, LLCSegmentName> partitionGroupIdToLatestSegment = new
HashMap<>();
Review Comment:
(minor) Partition group is an old term. Let's just stick with partition.
Same for javadoc
```suggestion
Map<Integer, LLCSegmentName> partitionIdToLatestSegment = new
HashMap<>();
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1481,6 +1471,81 @@ public void ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamCon
}
}
+ /// Fetches, from a read-only snapshot of the ideal state and outside the
ideal-state update lock, the stream
+ /// state needed by [#ensureAllPartitionsConsuming]: the latest
partition-group metadata (with start offsets) and,
+ /// when required, the smallest stream offset per partition.
+ ///
+ /// The smallest-offset fetch is a real stream round-trip, so it is only
performed when it can actually be used:
+ /// on a reset (`offsetCriteria != null`) or when at least one partition
currently lacks a CONSUMING segment and
+ /// may need a new one created. On a healthy table nothing is fetched and
`null` is returned for it, signalling
+ /// the repair pass to reuse the start offsets. When the criteria is
SMALLEST the start offsets already are the
+ /// smallest offsets, so it is likewise left `null`.
+ ///
+ /// Any temporary mutation of the shared `streamConfigs` offset criteria is
always restored, even on error.
+ @VisibleForTesting
+ PreFetchedOffsets preFetchOffsets(List<StreamConfig> streamConfigs, String
realtimeTableName,
+ IdealState snapshotIdealState, OffsetCriteria offsetCriteria) {
+ boolean offsetsHaveToChange = offsetCriteria != null;
+ List<PartitionGroupConsumptionStatus>
currentPartitionGroupConsumptionStatusList =
+ offsetsHaveToChange ? List.of()
+ // offsets from metadata are not valid anymore; fetch for all
partitions
+ : getPartitionGroupConsumptionStatusList(snapshotIdealState,
streamConfigs);
+ // FIXME: Right now, we assume topics are sharing same offset criteria
+ OffsetCriteria originalOffsetCriteria =
streamConfigs.get(0).getOffsetCriteria();
+ // For the periodic run, compute start offsets with SMALLEST so a newly
detected partition starts from the
+ // beginning; for a reset, use the requested criteria. Restored in the
finally below.
+ streamConfigs.forEach(streamConfig -> streamConfig.setOffsetCriteria(
+ offsetsHaveToChange ? offsetCriteria :
OffsetCriteria.SMALLEST_OFFSET_CRITERIA));
+ try {
+ List<StreamMetadata> streamMetadataList =
+ getNewStreamMetadataList(streamConfigs,
currentPartitionGroupConsumptionStatusList, snapshotIdealState);
+ Map<Integer, StreamPartitionMsgOffset> partitionIdToSmallestOffset =
null;
+ if (offsetCriteria == null ||
!offsetCriteria.equals(OffsetCriteria.SMALLEST_OFFSET_CRITERIA)) {
+ // Decide whether the smallest-offset stream fetch is needed from the
snapshot ideal state alone (no ZK
+ // metadata reads); only build the latest-segment ZK metadata map when
the fetch is actually required.
+ if (offsetsHaveToChange ||
anyPartitionNeedsSmallestOffset(snapshotIdealState)) {
+ partitionIdToSmallestOffset =
fetchPartitionGroupIdToSmallestOffset(streamConfigs, snapshotIdealState,
+ getLatestSegmentZKMetadataMap(realtimeTableName));
+ }
+ }
+ return new PreFetchedOffsets(streamMetadataList,
partitionIdToSmallestOffset);
+ } finally {
+ streamConfigs.forEach(streamConfig ->
streamConfig.setOffsetCriteria(originalOffsetCriteria));
+ }
+ }
+
+ /// Returns true if at least one partition's latest LLC segment in the ideal
state has no replica in the CONSUMING
+ /// state, i.e. it may need a new CONSUMING segment created (the only repair
path that consults the smallest stream
+ /// offset). Derived from the snapshot ideal state alone (no ZK metadata
reads), picking the latest segment per
+ /// partition the same way [#getPartitionGroupConsumptionStatusList] does.
+ private boolean anyPartitionNeedsSmallestOffset(IdealState idealState) {
+ Map<String, Map<String, String>> instanceStatesMap =
idealState.getRecord().getMapFields();
+ Map<Integer, LLCSegmentName> partitionGroupIdToLatestSegment =
+ getLatestLLCSegmentPerPartition(instanceStatesMap.keySet());
+ for (LLCSegmentName latestSegment :
partitionGroupIdToLatestSegment.values()) {
+ Map<String, String> instanceStateMap =
instanceStatesMap.get(latestSegment.getSegmentName());
+ if (instanceStateMap != null &&
!instanceStateMap.containsValue(SegmentStateModel.CONSUMING)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /// Holder for the stream state pre-fetched by [#preFetchOffsets] outside
the ideal-state update lock.
+ /// `_partitionIdToSmallestOffset` is null when the smallest offsets were
not fetched (see [#preFetchOffsets]).
+ @VisibleForTesting
+ static class PreFetchedOffsets {
Review Comment:
(minor) Can be simplified into a record class
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1448,28 +1448,34 @@ public void ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamCon
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
+
+ // Fetch all stream offsets BEFORE acquiring the ideal-state update lock.
On tables with many partitions these
+ // stream round-trips can take minutes; doing them here (against a
snapshot of the ideal state) keeps the
+ // per-table ideal-state lock hold-time proportional to the in-memory
ideal-state mutation, not to the offset
+ // I/O. The updater lambda below performs no stream I/O, so it is also
cheap to re-run on ZK CAS retries.
+ IdealState snapshotIdealState =
HelixHelper.getTableIdealState(_helixManager, realtimeTableName);
Review Comment:
(MAJOR) I'm still confused. `_streamMetadataList` is provided by the
`PreFetchedOffsets`, and I don't see it being updated during the retry. How
does it handle new segment being added?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
