HBASE-26384 Segment already flushed to hfile may still be remained in…

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java

@@ -567,12 +567,12 @@ private void pushPipelineToSnapshot() {
     boolean done = false;
     while (!done) {
       iterationsCnt++;
-      VersionedSegmentsList segments = pipeline.getVersionedList();
+      VersionedSegmentsList segments = getImmutableSegments();
       pushToSnapshot(segments.getStoreSegments());
       // swap can return false in case the pipeline was updated by ongoing compaction
       // and the version increase, the chance of it happenning is very low
       // In Swap: don't close segments (they are in snapshot now) and don't update the region size
-      done = pipeline.swap(segments, null, false, false);
+      done = swapPipelineWithNull(segments);
       if (iterationsCnt>2) {
         // practically it is impossible that this loop iterates more than two times
         // (because the compaction is stopped and none restarts it while in snapshot request),
@@ -585,6 +585,10 @@ private void pushPipelineToSnapshot() {
     }
   }
 
+  protected boolean swapPipelineWithNull(VersionedSegmentsList segments) {
+    return pipeline.swap(segments, null, false, false);
+  }
+
   private void pushToSnapshot(List<ImmutableSegment> segments) {
     if(segments.isEmpty()) return;
     if(segments.size() == 1 && !segments.get(0).isEmpty()) {

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java

@@ -18,7 +18,6 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
-import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
@@ -64,7 +63,16 @@ public class CompactionPipeline {
   private final LinkedList<ImmutableSegment> pipeline = new LinkedList<>();
   // The list is volatile to avoid reading a new allocated reference before the c'tor is executed
   private volatile LinkedList<ImmutableSegment> readOnlyCopy = new LinkedList<>();
-  // Version is volatile to ensure it is atomically read when not using a lock
+  /**
+   * <pre>
+   * Version is volatile to ensure it is atomically read when not using a lock.
+   * To indicate whether the suffix of pipleline changes：
+   * 1.for {@link CompactionPipeline#pushHead(MutableSegment)},new {@link ImmutableSegment} only
+   *   added at Head, {@link #version} not change.
+   * 2.for {@link CompactionPipeline#swap},{@link #version} increase.
+   * 3.for {@link CompactionPipeline#replaceAtIndex},{@link #version} increase.
+   * </pre>
+   */
   private volatile long version = 0;
 
   public CompactionPipeline(RegionServicesForStores region) {
@@ -95,7 +103,7 @@ public VersionedSegmentsList getVersionedList() {
 
   public VersionedSegmentsList getVersionedTail() {
     synchronized (pipeline){
-      List<ImmutableSegment> segmentList = new ArrayList<>();
+      LinkedList<ImmutableSegment> segmentList = new LinkedList<>();
       if(!pipeline.isEmpty()) {
         segmentList.add(0, pipeline.getLast());
       }
@@ -125,7 +133,7 @@ public boolean swap(VersionedSegmentsList versionedList, ImmutableSegment segmen
     if (versionedList.getVersion() != version) {
       return false;
     }
-    List<ImmutableSegment> suffix;
+    LinkedList<ImmutableSegment> suffix;
     synchronized (pipeline){
       if(versionedList.getVersion() != version) {
         return false;
@@ -290,10 +298,15 @@ public MemStoreSize getPipelineSize() {
     return memStoreSizing.getMemStoreSize();
   }
 
-  private void swapSuffix(List<? extends Segment> suffix, ImmutableSegment segment,
+  /**
+   * Must be called under the {@link CompactionPipeline#pipeline} Lock.
+   */
+  private void swapSuffix(LinkedList<? extends Segment> suffix, ImmutableSegment segment,
       boolean closeSegmentsInSuffix) {
-    pipeline.removeAll(suffix);
-    if(segment != null) pipeline.addLast(segment);
+    matchAndRemoveSuffixFromPipeline(suffix);
+    if (segment != null) {
+      pipeline.addLast(segment);
+    }
     // During index merge we won't be closing the segments undergoing the merge. Segment#close()
     // will release the MSLAB chunks to pool. But in case of index merge there wont be any data copy
     // from old MSLABs. So the new cells in new segment also refers to same chunks. In case of data
@@ -307,6 +320,41 @@ private void swapSuffix(List<? extends Segment> suffix, ImmutableSegment segment
     }
   }
 
+  /**
+   * Checking that the {@link Segment}s in suffix input parameter is same as the {@link Segment}s in
+   * {@link CompactionPipeline#pipeline} one by one from the last element to the first element of
+   * suffix. If matched, remove suffix from {@link CompactionPipeline#pipeline}. <br/>
+   * Must be called under the {@link CompactionPipeline#pipeline} Lock.
+   */
+  private void matchAndRemoveSuffixFromPipeline(LinkedList<? extends Segment> suffix) {
+    if (suffix.isEmpty()) {
+      return;
+    }
+    if (pipeline.size() < suffix.size()) {
+      throw new IllegalStateException(
+          "CODE-BUG:pipleine size:[" + pipeline.size() + "],suffix size:[" + suffix.size()
+              + "],pipeline size must greater than or equals suffix size");
+    }
+
+    Iterator<? extends Segment> suffixIterator = suffix.descendingIterator();
+    Iterator<? extends Segment> pipelineIterator = pipeline.descendingIterator();
+    int count = 0;
+    while (suffixIterator.hasNext()) {
+      Segment suffixSegment = suffixIterator.next();
+      Segment pipelineSegment = pipelineIterator.next();
+      if (suffixSegment != pipelineSegment) {
+        throw new IllegalStateException("CODE-BUG:suffix last:[" + count + "]" + suffixSegment
+            + " is not pipleline segment:[" + pipelineSegment + "]");
+      }
+      count++;
+    }
+
+    for (int index = 1; index <= count; index++) {
+      pipeline.pollLast();
+    }
+
+  }
+
   // replacing one segment in the pipeline with a new one exactly at the same index
   // need to be called only within synchronized block
   @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="VO_VOLATILE_INCREMENT",

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/VersionedSegmentsList.java

@@ -18,7 +18,7 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
-import java.util.List;
+import java.util.LinkedList;
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**
@@ -34,15 +34,15 @@
 @InterfaceAudience.Private
 public class VersionedSegmentsList {
 
-  private final List<ImmutableSegment> storeSegments;
+  private final LinkedList<ImmutableSegment> storeSegments;
   private final long version;
 
-  public VersionedSegmentsList(List<ImmutableSegment> storeSegments, long version) {
+  public VersionedSegmentsList(LinkedList<ImmutableSegment> storeSegments, long version) {
     this.storeSegments = storeSegments;
     this.version = version;
   }
 
-  public List<ImmutableSegment> getStoreSegments() {
+  public LinkedList<ImmutableSegment> getStoreSegments() {
     return storeSegments;
   }