Copilot commented on code in PR #19170:
URL: https://github.com/apache/pinot/pull/19170#discussion_r3762580946
##########
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);
+ if (snapshotIdealState == null) {
+ LOGGER.warn("Cannot find ideal state for table: {}, skipping
ensureAllPartitionsConsuming", realtimeTableName);
+ return;
+ }
+ if (!snapshotIdealState.isEnabled() || isTablePaused(snapshotIdealState)) {
+ LOGGER.info("Skipping LLC segments validation for table: {},
isTableEnabled: {}, isTablePaused: {}",
+ realtimeTableName, snapshotIdealState.isEnabled(),
isTablePaused(snapshotIdealState));
+ return;
+ }
+
try {
+ PreFetchedOffsets preFetchedOffsets =
+ preFetchOffsets(streamConfigs, realtimeTableName,
snapshotIdealState, offsetCriteria);
+
HelixHelper.updateIdealState(_helixManager, realtimeTableName,
idealState -> {
assert idealState != null;
boolean isTableEnabled = idealState.isEnabled();
boolean isTablePaused = isTablePaused(idealState);
- boolean offsetsHaveToChange = offsetCriteria != null;
if (isTableEnabled && !isTablePaused) {
- List<PartitionGroupConsumptionStatus>
currentPartitionGroupConsumptionStatusList =
- offsetsHaveToChange ? List.of()
- // offsets from metadata are not valid anymore; fetch for
all partitions
- : getPartitionGroupConsumptionStatusList(idealState,
streamConfigs);
- // FIXME: Right now, we assume topics are sharing same offset
criteria
- OffsetCriteria originalOffsetCriteria =
streamConfigs.get(0).getOffsetCriteria();
- // Read the smallest offset when a new partition is detected
- streamConfigs.stream()
- .forEach(streamConfig -> streamConfig.setOffsetCriteria(
- offsetsHaveToChange ? offsetCriteria :
OffsetCriteria.SMALLEST_OFFSET_CRITERIA));
- List<StreamMetadata> streamMetadataList =
- getNewStreamMetadataList(streamConfigs,
currentPartitionGroupConsumptionStatusList, idealState);
- streamConfigs.stream().forEach(streamConfig ->
streamConfig.setOffsetCriteria(originalOffsetCriteria));
- return ensureAllPartitionsConsuming(tableConfig, streamConfigs,
idealState, streamMetadataList,
- offsetCriteria);
+ return ensureAllPartitionsConsuming(tableConfig, streamConfigs,
idealState,
+ preFetchedOffsets._streamMetadataList, offsetCriteria,
Review Comment:
The fresh-state recheck only covers a whole-table pause. A partial topic
pause updates `PauseState.indexOfInactiveTopics` while `isTablePaused()`
remains false, so if that update lands after the snapshot, this lambda still
applies metadata fetched for the now-paused topic. It can repair that topic or
create a newly discovered partition after the pause, reintroducing CONSUMING
segments that the pause operation intentionally stopped. Revalidate per-topic
pause state against this fresh `idealState` before both repair and
new-partition creation (and add the pause-between-prefetch-and-update
interleaving test).
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1481,6 +1487,78 @@ 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)) {
+ Map<Integer, SegmentZKMetadata> latestSegmentZKMetadataMap =
getLatestSegmentZKMetadataMap(realtimeTableName);
+ if (offsetsHaveToChange ||
anyPartitionNeedsSmallestOffset(snapshotIdealState,
latestSegmentZKMetadataMap)) {
+ partitionIdToSmallestOffset =
+ fetchPartitionGroupIdToSmallestOffset(streamConfigs,
snapshotIdealState, latestSegmentZKMetadataMap);
Review Comment:
This eagerly performs another full latest-metadata scan merely to decide
whether smallest offsets are needed. `getPartitionGroupConsumptionStatusList()`
above already reads the latest ZK metadata once per partition, and the updater
reads it again at line 1817; this call adds a third O(partitions) set of
PropertyStore reads on every healthy validation cycle (about 1,024 extra reads
for the motivating table). Determine the gate from the snapshot's latest
IdealState segment names/current statuses, and only build the ZK metadata map
when the smallest-offset fetch is actually required.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java:
##########
@@ -2494,8 +2625,11 @@ public void setUpNewTable() {
}
public void ensureAllPartitionsConsuming() {
- ensureAllPartitionsConsuming(_tableConfig, _streamConfigs, _idealState,
- getNewStreamMetadataList(_streamConfigs, List.of(),
mock(IdealState.class)), null);
+ // Mirror the production flow: pre-fetch offsets (gated) outside the
ideal-state update, then pass them into the
+ // package-private repair method.
+ PreFetchedOffsets preFetchedOffsets = preFetchOffsets(_streamConfigs,
REALTIME_TABLE_NAME, _idealState, null);
+ ensureAllPartitionsConsuming(_tableConfig, _streamConfigs, _idealState,
preFetchedOffsets._streamMetadataList,
+ null, preFetchedOffsets._partitionIdToSmallestOffset);
Review Comment:
This test helper bypasses the changed public method and calls prefetch and
repair sequentially itself, so none of the tests verifies the core regression:
stream metadata calls must occur before `HelixHelper.updateIdealState`, and CAS
retries must not repeat them. Add a test around the public
`ensureAllPartitionsConsuming` path with a controlled Helix accessor/updater
that records call order and retries the updater; assert each stream fetch
occurs once and outside the updater.
--
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]
