shounakmk219 commented on code in PR #19170:
URL: https://github.com/apache/pinot/pull/19170#discussion_r3793991054
##########
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:
topic level pause was not handled before so skipping it in this refactor as
well
--
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]
