Fix the race condition in realtime text index refresh thread (#6858)

[GSharayu]

The refresh thread maintains a single global queue of realtime text index readers across all consuming segments across all tables.

A reader is polled, refreshed and supposed to be added back to the queue so that it can be refreshed again in the next cycle.
There is a race condition: If we find the queue empty, but something gets added before we lock the mutex, the thread signal will be lost and it will be in await mode until next segment signals.

Set of steps for race condition

1. Thread finds queue empty and enters the while check on https://github.com/apache/incubator-pinot/blob/master/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneIndexReaderRefreshThread.java#L69
2. Meanwhile another thread grabs lock at https://github.com/apache/incubator-pinot/blob/master/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneIndexRefreshState.java#L86
3. The current thread adds the segment in queue and signals the condition variable (right now its not hold)
4. The lock after released from step 3 will be grabbed by thread in step 1
   It will then await for signal on condition variable(Which is bug!) and might have to wait until some other segment signals it.

here are several ways to solve the above problem

1. One easy way to fix this is to check again if the queue is empty right before the condition variable awaits.
2. Another similar way is to use a variable and set it during the segment addition and a similar if condition check(as 1) for the variable before condition variable awaits.
3. The approach used in above PR
   i. When the thread enters https://github.com/apache/incubator-pinot/blob/master/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneIndexReaderRefreshThread.java#L68, it will try to acquire the lock.
   ii. If it does acquire the lock, it will check if the queue is empty or not. Depending on state of queue, it will either skip the while loop or enter the while loop and await for the signal (and release the lock while await())
   iii. If the thread at https://github.com/apache/incubator-pinot/blob/master/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/invertedindex/RealtimeLuceneIndexRefreshState.java#L86 acquires lock first, then it will add the segment to queue and signal any condition variable if it awaits.
   iv. In step 3, after segment is added to queue, segment thread will release the lock. and when background refresh thread now tries to acquire the lock, gets the lock. It then will check for while condition(if the queue is empty) which will be false after step 3 and never wait. In case any problem occurs within while loop, it being in try catch{} block, the lock acquired by this background refresh thread will be released in finally{}.

Essentially, the mutex ensures that only a single(background refresh thread/segment thread) thread updates/check the queue at any given moment.

Issue #6858

cc @siddharthteotia @mcvsubbu

[codecov-commenter]

Codecov Report

Merging #6990 (1e9456b) into master (6388c77) will increase coverage by 0.13%.
The diff coverage is 57.14%.

❗ Current head 1e9456b differs from pull request most recent head 89bd89d. Consider uploading reports for the commit 89bd89d to get more accurate results

@@             Coverage Diff              @@
##             master    #6990      +/-   ##
============================================
+ Coverage     73.31%   73.45%   +0.13%     
  Complexity       12       12              
============================================
  Files          1441     1441              
  Lines         71432    71432              
  Branches      10351    10351              
============================================
+ Hits          52369    52469     +100     
+ Misses        15542    15444      -98     
+ Partials       3521     3519       -2     

Flag	Coverage Δ
integration	42.06% <57.14%> (+0.17%)	⬆️
unittests	65.53% <0.00%> (+<0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...dindex/RealtimeLuceneIndexReaderRefreshThread.java	71.73% <57.14%> (ø)
...operator/filter/RangeIndexBasedFilterOperator.java	40.42% <0.00%> (-4.26%)	⬇️
.../java/org/apache/pinot/spi/data/TimeFieldSpec.java	88.63% <0.00%> (-2.28%)	⬇️
...mpl/dictionary/DoubleOffHeapMutableDictionary.java	55.31% <0.00%> (-2.13%)	⬇️
...impl/dictionary/FloatOffHeapMutableDictionary.java	76.59% <0.00%> (-1.07%)	⬇️
...manager/realtime/LLRealtimeSegmentDataManager.java	73.06% <0.00%> (-0.86%)	⬇️
...e/impl/dictionary/LongOnHeapMutableDictionary.java	69.87% <0.00%> (ø)
...ache/pinot/common/metadata/ZKMetadataProvider.java	77.95% <0.00%> (+0.53%)	⬆️
...a/org/apache/pinot/core/common/DataBlockCache.java	96.96% <0.00%> (+0.75%)	⬆️
.../controller/helix/core/SegmentDeletionManager.java	77.86% <0.00%> (+0.81%)	⬆️
... and 20 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 6388c77...89bd89d. Read the comment docs.

[siddharthteotia]

LGTM.

There is another thing that needs to be fixed w.r.t setting the volatile stopped variable. Since that is not directly related to this change, we can do in a follow-up.