KKcorps opened a new pull request, #19433:
URL: https://github.com/apache/pinot/pull/19433
## TL;DR
`RealtimeSegmentDataManager.doOffload()` released the consumer semaphore two
calls before it removed
the segment from the upsert and dedup metadata managers. For partial-upsert
tables in `PROTECTED`
consistency mode that removal is where primary keys are reverted to their
previous locations, so the
next consuming segment of the partition could start replaying while keys
still pointed at the mutable
segment being offloaded. This PR runs the metadata removal first and
releases the semaphore in a
`finally` block.
## The problem
On the normal commit path this ordering already holds: `replaceSegment()`
reverts the keys before
`offload()` runs, so the semaphore release comes after. The
offload-without-replace path
(CONSUMING to OFFLINE, CONSUMING to DROPPED) is the one that was wrong.
```mermaid
flowchart LR
subgraph Before["❌ Old order"]
A1[stop consumer] --> A2[release semaphore] --> A3[remove upsert
metadata / revert]
A2 -.-> A4[next segment starts replaying]
A4 -.-> A5[merges against un-reverted keys]
end
subgraph After["✅ New order"]
B1[stop consumer] --> B2[remove upsert metadata / revert] --> B3[release
semaphore]
B3 --> B4[next segment starts replaying]
B4 --> B5[merges against reverted keys]
end
```
What went wrong per key when the successor won the race:
- `doUpdateRecord` found the key still owned by the mutable segment with an
equal comparison value,
passed the out-of-order guard, and merged the already-merged row with the
same update again. For
`INCREMENT`, `APPEND` and `UNION` strategies that is a double count.
- `doAddRecord` wrote no undo entry because the current owner was a mutable
segment.
- When the revert reached that key it saw another mutable owner, logged the
"consumption is
occurring concurrently with segment replacement" warning, and skipped it.
This needed a successor consumer already blocked on the semaphore for the
same partition on the
same server, so it is rare. It is also cheap to close.
## Scope
This closes the gap only for consumption policies that hold the semaphore
until offload, which is
`DISALLOW_ALWAYS`, the default for partial upsert without pauseless.
`ALLOW_DURING_BUILD_ONLY` (the
pauseless default) and `ALLOW_ALWAYS` release the semaphore earlier, from
`buildSegmentInternal()`
or `downloadSegmentAndReplace()`, by design. Overlap under those policies is
a separate problem and
is not addressed here.
## The approach
1. `stop()` the consumer thread, unchanged.
2. Call `_realtimeSegment.offload()`, which calls
`PartitionUpsertMetadataManager.removeSegment()`
and `PartitionDedupMetadataManager.removeSegment()`.
3. In a `finally` block, `closeStreamConsumer()` (this releases the
semaphore) and
`cleanupMetrics()`. The `finally` matters: if the revert throws, the
semaphore must still be
released or the partition can never consume again on this server.
## Flow
```mermaid
sequenceDiagram
participant H as Helix thread
participant D as RealtimeSegmentDataManager
participant M as MutableSegmentImpl
participant U as PartitionUpsertMetadataManager
participant C as ConsumerCoordinator
participant N as Next consuming segment
H->>D: doOffload()
D->>D: stop()
D->>M: offload()
M->>U: removeSegment(segment)
U-->>M: keys reverted to previous locations
D->>C: release() (in finally)
C-->>N: semaphore acquired
N->>U: addRecord / updateRecord
Note over N,U: keys already point at committed segments
```
## Testing
Two unit tests in `RealtimeSegmentDataManagerTest`, using a Mockito spy on
the mutable segment:
- `testOffloadRemovesSegmentMetadataBeforeReleasingConsumerSemaphore`:
acquires the coordinator's
semaphore, records `availablePermits()` inside
`MutableSegmentImpl.offload()`, and asserts it is 0
at that point and 1 after `offload()` returns.
- `testOffloadReleasesConsumerSemaphoreWhenMetadataRemovalFails`: makes
`offload()` throw and asserts
the exception propagates and the semaphore is still released.
Both tests use a fresh `ConsumerCoordinator`, because other tests in the
class release the shared
semaphore without acquiring it.
## Base
`master`. No config, metric, or wire-format changes.
## Labels
`bugfix`
--
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]
