Jackie-Jiang commented on code in PR #20106:
URL: https://github.com/apache/pinot/pull/20106#discussion_r4448373104
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -235,43 +242,68 @@ 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()) {
Review Comment:
Addressed in 24e449d42a. The shared `requiresHistoryForConsumption()` policy
now uses `UpsertContext.isTableTypeInconsistentDuringConsumption()`, covering
partial upsert and full upsert with effective `dropOutOfOrderRecord` or
`outOfOrderRecordColumn`, in addition to dedup. This uses the effective
context, so drop/mark options disabled by SYNC/SNAPSHOT do not add a history
gate. Added coverage for both options, consistency modes, and ordering
enabled/disabled.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/ConsumerCoordinator.java:
##########
@@ -116,14 +121,65 @@ public void acquire(LLCSegmentName llcSegmentName)
}
}
+ /// Waits for required history before acquiring the consumer permit. Local
permit releases wake all waiters;
+ /// Helix assignment and state changes still use the periodic check.
Readiness can perform ZooKeeper reads, so it
+ /// must run outside the monitor. The version prevents losing a release
between the readiness check and the wait.
+ /// @return whether readiness passed, or false if consumption was stopped
+ boolean waitUntilReadyToConsume(int partitionGroupId, BooleanSupplier
isReadyToConsume, BooleanSupplier shouldStop)
Review Comment:
Agreed: the complete history check subsumes predecessor registration for
these tables. Addressed in 24e449d42a by moving readiness into
`ConsumerCoordinator.acquire()` and making the two ordering policies exclusive.
Tables requiring history use only the history wait; other tables retain
predecessor registration when configured. The history gate still applies when
`enforceConsumptionInOrder` is false. The caller now makes one admission call,
and the coordinator also owns cancellation and permit handoff.
##########
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:
Addressed in 7d0118597d and retained in 24e449d42a. Ordinary rollover marks
the replacement immutable manager ready only after metadata replacement
completes, before predecessor offload releases the permit. The successor can
therefore recognize local completion while CurrentState is still CONSUMING.
Release wakes the readiness wait, with a version counter preventing a missed
wakeup; pauseless rollover uses the explicit unsealed-release marker. The
threaded ordinary-rollover regression covers this handoff and verifies that
missing older history still blocks. Helix-only changes retain the periodic
fallback.
##########
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:
Addressed in 7d0118597d with a shared immutable table snapshot for negative
readiness checks, reused for up to five seconds during steady waiting. Local
metadata readiness is re-evaluated against that snapshot. A cached positive
result requires a fresh assignment/history check, and a new consumer refreshes
a snapshot that predates its creation. In 24e449d42a, after the initial
unsuccessful check, a waiter also skips repeated history reads while another
consumer holds the permit and rechecks on release. New-consumer checks,
positive verification, and failed refresh retries can still cause additional
reads; runtime performance remains unbenchmarked.
--
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]
