xiangfu0 commented on code in PR #20106:
URL: https://github.com/apache/pinot/pull/20106#discussion_r4429247412
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/tablestate/TableStateUtils.java:
##########
@@ -82,12 +84,56 @@ public static List<String>
getSegmentsInGivenStateForThisInstance(HelixManager h
public static boolean isAllSegmentsLoaded(HelixManager helixManager, String
tableNameWithType) {
List<String> onlineSegments =
getSegmentsInGivenStateForThisInstance(helixManager,
tableNameWithType, SegmentStateModel.ONLINE);
+ return areSegmentsLoaded(helixManager, tableNameWithType, onlineSegments);
+ }
+
+ /// Checks the current assignment before a consumer starts. Earlier ONLINE
segments of the same partition must be
+ /// loaded. Non-LLC ONLINE segments are conservatively included because they
may hold base rows for the partition.
+ /// The consumer itself and its successors are excluded so that they do not
wait for their own completion.
+ ///
+ /// @param isLocallyReady identifies segments whose metadata is already
ready locally despite a pending ONLINE
+ /// transition, such as a predecessor that finished
consumption and released its permit
+ public static boolean isReadyToConsume(HelixManager helixManager, String
tableNameWithType, LLCSegmentName consumer,
+ Predicate<String> isLocallyReady) {
+ HelixDataAccessor dataAccessor = helixManager.getHelixDataAccessor();
+ IdealState idealState =
dataAccessor.getProperty(dataAccessor.keyBuilder().idealStates(tableNameWithType));
+ if (idealState == null) {
+ LOGGER.warn("Failed to find ideal state for table: {}",
tableNameWithType);
+ return false;
+ }
+ String instanceName = helixManager.getInstanceName();
+ Map<String, String> consumerAssignment =
idealState.getInstanceStateMap(consumer.getSegmentName());
+ String consumerState = consumerAssignment != null ?
consumerAssignment.get(instanceName) : null;
+ if (!SegmentStateModel.CONSUMING.equals(consumerState) &&
!SegmentStateModel.ONLINE.equals(consumerState)) {
+ return false;
+ }
+
+ List<String> requiredSegments = new ArrayList<>();
+ for (Map.Entry<String, Map<String, String>> entry :
idealState.getRecord().getMapFields().entrySet()) {
+ if
(!SegmentStateModel.ONLINE.equals(entry.getValue().get(instanceName))) {
+ continue;
+ }
+ String segmentName = entry.getKey();
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null && (llcSegmentName.getPartitionGroupId() !=
consumer.getPartitionGroupId()
+ || llcSegmentName.getSequenceNumber() >=
consumer.getSequenceNumber())) {
+ continue;
+ }
+ if (!isLocallyReady.test(segmentName)) {
+ requiredSegments.add(segmentName);
Review Comment:
**MAJOR (needs owner decision): an ERROR segment can halt ingestion
indefinitely.** `areSegmentsLoaded` returns `false` whenever a required segment
is in ERROR. This check now runs at every consumer start, not just once at
startup. So one old segment of the partition in ERROR stops that replica's
consumption until someone resets it. Because non-LLC segments are required for
every partition, an uploaded segment in ERROR stops all partitions. The only
signal is a log line every 5s. Stopping may be the right call for partial
upsert, but please confirm the intent and add a gauge/meter so the stall can be
alerted on.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -235,43 +235,38 @@ protected void doInit() {
_enforceConsumptionInOrder = isEnforceConsumptionInOrder();
_streamMetadataProviderCache = getStreamMetadataProviderCache();
- // For dedup and partial-upsert, need to wait for all segments loaded
before starting consuming data
- BooleanSupplier readyForDedupOrPartialUpsert;
- if (isDedupEnabled() || isPartialUpsertEnabled()) {
- readyForDedupOrPartialUpsert = new BooleanSupplier() {
- volatile boolean _allSegmentsLoaded;
- long _lastCheckTimeMs;
-
- @Override
- public boolean getAsBoolean() {
- if (_allSegmentsLoaded) {
- return true;
- } else {
- synchronized (this) {
- if (_allSegmentsLoaded) {
- return true;
- }
- long currentTimeMs = System.currentTimeMillis();
- if (currentTimeMs - _lastCheckTimeMs <=
READY_TO_CONSUME_DATA_CHECK_INTERVAL_MS) {
- return false;
- }
- _lastCheckTimeMs = currentTimeMs;
- _allSegmentsLoaded =
TableStateUtils.isAllSegmentsLoaded(_helixManager, _tableNameWithType);
- return _allSegmentsLoaded;
- }
- }
- }
- };
- } else {
- readyForDedupOrPartialUpsert = () -> true;
- }
- _isTableReadyToConsumeData = () ->
readyForDedupOrPartialUpsert.getAsBoolean()
- && _isServerReadyToConsumeData.getAsBoolean();
_serverIngestionOomProtectionManager = new
ServerIngestionOomProtectionManager(
() -> getCachedTableConfigAndSchema().getLeft(), () ->
isUpsertEnabled() || isDedupEnabled(),
_serverIngestionOomProtectionThrottleState);
}
+ /// Checks the assigned history again for each consumer, including
partitions added after table initialization.
+ /// The consumer thread polls this supplier before acquiring its permit, so
segment loading is never blocked by it.
+ @VisibleForTesting
+ BooleanSupplier createReadyToConsumeDataSupplier(LLCSegmentName
llcSegmentName) {
+ if (!isDedupEnabled() && !isPartialUpsertEnabled()) {
+ return _isServerReadyToConsumeData;
+ }
+ return () -> {
Review Comment:
**MAJOR: up to 5s stall on every rollover.** This check now runs before
`acquire()` for every consumer. Under `DISALLOW_ALWAYS` (the default for
non-pauseless partial upsert and dedup), `n-1` never sets
`hasReleasedConsumerSemaphore`, so `n` waits until `n-1`'s CURRENTSTATE is
ONLINE. It then only notices on the next fixed 5s `Thread.sleep` tick. Before
this change, it blocked in `acquire()` and woke as soon as the permit was
released. Pauseless has the same problem whenever the first check runs before
`n-1` reaches its build. My probe confirmed this: with `n-1` already replaced
locally (immutable segment, metadata loaded) but CURRENTSTATE still CONSUMING,
the supplier returns `false`. Could the local coordinator handoff count for the
immediate predecessor, or could the wait wake on release instead of the 5s
timer?
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -235,43 +235,38 @@ protected void doInit() {
_enforceConsumptionInOrder = isEnforceConsumptionInOrder();
_streamMetadataProviderCache = getStreamMetadataProviderCache();
- // For dedup and partial-upsert, need to wait for all segments loaded
before starting consuming data
- BooleanSupplier readyForDedupOrPartialUpsert;
- if (isDedupEnabled() || isPartialUpsertEnabled()) {
- readyForDedupOrPartialUpsert = new BooleanSupplier() {
- volatile boolean _allSegmentsLoaded;
- long _lastCheckTimeMs;
-
- @Override
- public boolean getAsBoolean() {
- if (_allSegmentsLoaded) {
- return true;
- } else {
- synchronized (this) {
- if (_allSegmentsLoaded) {
- return true;
- }
- long currentTimeMs = System.currentTimeMillis();
- if (currentTimeMs - _lastCheckTimeMs <=
READY_TO_CONSUME_DATA_CHECK_INTERVAL_MS) {
- return false;
- }
- _lastCheckTimeMs = currentTimeMs;
- _allSegmentsLoaded =
TableStateUtils.isAllSegmentsLoaded(_helixManager, _tableNameWithType);
- return _allSegmentsLoaded;
- }
- }
- }
- };
- } else {
- readyForDedupOrPartialUpsert = () -> true;
- }
- _isTableReadyToConsumeData = () ->
readyForDedupOrPartialUpsert.getAsBoolean()
- && _isServerReadyToConsumeData.getAsBoolean();
_serverIngestionOomProtectionManager = new
ServerIngestionOomProtectionManager(
() -> getCachedTableConfigAndSchema().getLeft(), () ->
isUpsertEnabled() || isDedupEnabled(),
_serverIngestionOomProtectionThrottleState);
}
+ /// Checks the assigned history again for each consumer, including
partitions added after table initialization.
+ /// The consumer thread polls this supplier before acquiring its permit, so
segment loading is never blocked by it.
+ @VisibleForTesting
+ BooleanSupplier createReadyToConsumeDataSupplier(LLCSegmentName
llcSegmentName) {
+ if (!isDedupEnabled() && !isPartialUpsertEnabled()) {
+ return _isServerReadyToConsumeData;
+ }
+ return () -> {
+ if (!_isServerReadyToConsumeData.getAsBoolean()) {
+ return false;
+ }
+ try {
+ return TableStateUtils.isReadyToConsume(_helixManager,
_tableNameWithType, llcSegmentName, segmentName -> {
Review Comment:
**MAJOR: ZK reads now multiply by partition count.** The old supplier was
shared, `synchronized`, and throttled by `_lastCheckTimeMs`, so a table did at
most one check per 5s. Now each waiting consumer reads the full IdealState plus
LiveInstance and CurrentState on every poll. My probe with 64 waiting
partitions measured 64 IdealState reads and 192 ZK reads per polling round. For
large upsert tables, a server restart or rebalance multiplies ZK traffic by the
number of partitions. Could consumers share a table-level IS/CS snapshot,
cached for the interval, and each evaluate their own partition against it?
--
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]
