noob-se7en commented on code in PR #19170:
URL: https://github.com/apache/pinot/pull/19170#discussion_r3871596809
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1481,6 +1473,76 @@ 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());
Review Comment:
this derives the latest segment from IS names but the repair loop derives it
from `getLLCSegments` (property store), so a CONSUMING segment in IS with no ZK
metadata (the orphan `IdealStateGroupCommit`'s cancellation comment describes)
keeps the gate false on every run while the loop keeps deferring that
partition, and unlike pre-PR nothing heals it (`repairSegmentsInErrorState`
skips metadata-less segments). gating on
`getLatestLLCSegmentPerPartition(getLLCSegments(realtimeTableName))` against
the snapshot IS is one `getChildNames` call and matches the loop exactly.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1745,10 +1808,22 @@ IdealState ensureAllPartitionsConsuming(TableConfig
tableConfig, List<StreamConf
partitionIdToStartOffset.put(metadata.getPartitionGroupId(),
metadata.getStartOffset());
}
}
- // Create a map from partition id to the smallest stream offset
- Map<Integer, StreamPartitionMsgOffset> partitionIdToSmallestOffset = null;
- if (offsetCriteria != null &&
offsetCriteria.equals(OffsetCriteria.SMALLEST_OFFSET_CRITERIA)) {
+ // Map from partition id to the smallest stream offset, pre-fetched
outside the ideal-state lock (see
+ // preFetchOffsets). Three cases:
+ // - non-null: the fetched map. A partition absent from it has reached
end of life.
+ // - null with SMALLEST offset criteria: the start offsets computed
above already are the smallest offsets, so
+ // reuse them (they were fetched with SMALLEST for every partition).
+ // - null otherwise: the lock-free snapshot gate saw no partition
needing a new CONSUMING segment, so the
+ // smallest offsets were not fetched. Start offsets are NOT the
stream-smallest in this case, so they must
+ // not be substituted; a partition that turns out to need a new
segment now (it started needing repair after
+ // the snapshot) is deferred to the next validation run below.
Review Comment:
"next validation run" is one RVM period, 1h by default, not the 15 min in
the description, and the window that lands a partition here is the whole
pre-fetch. pre-PR a last replica going OFFLINE mid-fetch was still repaired in
the same run via the CAS retry; worth a forced re-run when the lambda defers,
or at least the 1h in the description.
--
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]
