[Bug] Inconsistent data in partial upsert table

[rohityadav1993]

bug
We are observing that during segment commits, there is a situation in the new segment where instead of merging the row for an existing PK it is getting added as a new row, loosing all the mergers that happened. This is happening only in one replica thus causing inconsistent result to be returned for the same query executed mulitple times.

Build:
Pinot version: 1.0
Included patches to attempt a fix:
#12395
#12241
#12105

Possible know factors to reproduce:

1. High ingestion scale: 15k msg/sec
2. Use a constant value for comparisonColumn
3. Any partial upsert merger, easier to observe in append list.

Observation where hostA is affected:

select test_list, $segmentName, $docId from rta_test_table 
where pk = '2dc705af-b8e9-4e55-a483-1cc64a7002e7' 
-- and $hostName = 'hostA'
and $hostName = 'hostB'
order by $segmentName, $docId
limit 100
option(skipUpsert=true)

partialupsertkeysnotreplaced metric for the affected table:

diff:

[Jackie-Jiang]

cc @klsince

[tibrewalpratik17]

One potential root-cause for this might be releasing the _partitionGroupConsumerSemaphore before the segment actually came ONLINE. This can result in starting the consumption on new segment.

For example:

• Segment S1 was in CONSUMING state and is now getting moved to ONLINE.

• Segment S2 is the new segment which is waiting for acquiring the _partitionGroupConsumerSemaphore --

  pinot/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

  Line 1563 in a385e28

  _partitionGroupConsumerSemaphore.acquire();

• Segment S1 starts building the same segment and releases _partitionGroupConsumerSemaphore

  pinot/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

  Lines 964 to 965 in a385e28

  	SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) {
  	closeStreamConsumers();

  or it issues a downloadSegmentAndReplace step again releasing the semaphore first

  pinot/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java

  Lines 1320 to 1322 in a385e28

  	protected void downloadSegmentAndReplace(SegmentZKMetadata segmentZKMetadata)
  	throws Exception {
  	closeStreamConsumers();

  In both the scenarios, S2 will start the consumption and in another thread, S1 will be getting built.

• Now in this particular issue reported we might have issue with downloadSegmentAndReplace. We were observing is that the merge function in S2 was treating the incoming record as a new record and not picking up the changes already came in S1. This might be because the records never came in S1 and the consumption was stopped because it never caught up to the committing offset. If you see the ingestion throughput is pretty high here (15k msgs/sec) and all the records for a given key is coming in a very small space of time. So if S1 did not catch up to the first offset of a given key K1 in one of the replicas and that replica starts getting replaced, S2 in the same replica will treat it as a new key and persist it. And now S1 will report the keys as not_replaced which we were seeing in our metrics as well.

Question to @Jackie-Jiang and @klsince , can we move the closeStreamConsumers call after building the consuming segment? Do you know why we went with this logic in the first place?

Note: This will not affect realtime and full-upsert cases to a lot extent but becomes very critical for partial-upsert cases where replica will diverge over time.

[Jackie-Jiang]

We release the lock before segment is fully replaced to shorten the lag of segment commit.
The analysis is correct. For partial upsert case, we need to ensure all the segments are loaded before starting the next consuming segment. We do have this check when server starts, but not at the time when segment is committed

[rohityadav1993]

I am trying this out by moving all the closeStreamConsumers() calls after the replace segment logic. Will post the results.

[rohityadav1993]

I don't see any inconsistency issues after moving the logic of releasing partition consumer semaphore(_partitionGroupConsumerSemaphore) in closeStreamConsumers() after replace is complete.
There is also an indication that not just downloadAndReplace() flow is affected as we see the inconsistent segments even when there is no peer download and consuming segment is built and replaced.

To ensure data accuracy, we will have to ensure to release _partitionGroupConsumerSemaphore only after replace is replace is completed. The side effect is stopped ingestion during segment commit. On our test table this has resulted in a jump of ingestion lag from ~5 minutes to ~15 minutes
Before:

After:

@Jackie-Jiang , @klsince , do you have a suggestion how we should go about this:

1. For partial upserts, always release semaphore after replace
2. A new table config to release semaphore after replace e.g. pausePartitionConsumptionDuringReplace: true

_partitionGroupConsumerSemaphore acquires in RealtimeSegmentDataManager's constructor, so it would stay stuck in the constructor until it is released after replace.

Test setup:
Select rows in a time range and re-query the same rows and compare. Carry this for multiple time ranges and around the time when segment commit happens.

[rohityadav1993]

Fixed in #13256.