diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java index b5eb566..3d1ad37 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java @@ -2348,10 +2348,31 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi myseqid); } flushOpSeqId = getNextSequenceId(wal); - // Back up 1, minus 1 from oldest sequence id in memstore to get last 'flushed' edit - flushedSeqId = - earliestUnflushedSequenceIdForTheRegion.longValue() == HConstants.NO_SEQNUM? - flushOpSeqId: earliestUnflushedSequenceIdForTheRegion.longValue() - 1; + if (earliestUnflushedSequenceIdForTheRegion.longValue() == HConstants.NO_SEQNUM) { + flushedSeqId = flushOpSeqId; + // Since append will be published to RingBuffer and handled asynchronously, it's possible + // that one append (say append-X) of the region handled by RingBufferEventHandler between + // startCacheFlush and getNextSequenceId, and reset FSHLog#oldestUnflushedStoreSequenceIds + // which we just cleared in #startCacheFlush. This might disturb ServerManager#flushedSequenceIdByRegion + // like shown below (assume region-A has two CF: cfA and cfB) + // + // 1. flush-A runs to startCacheFlush and it will flush both cfA and cfB, + // oldestUnflushedStoreSequenceIds of regionA got cleared + // 2. append-X on cfB handled by RingBufferEventHandler, oldestUnflushedStoreSequenceIds set to 10, for example + // 3. flush-A runs to getNextSequenceId and returned 11 + // 4. ServerManager#flushedSequenceIdByRegion for regionA set to 11 + // 5. flush-A finishes + // 6. flush-B starts and only flush cfA, getNextSequenceId returned 10, and flushedSeqId + // will return 9, and cause warning in ServerManager + // + // Since this append-X will also got flushed, we should clear the + // oldestUnflushedStoreSequenceIds again to make sure we won't disturb + // ServerManager#flushedSequenceIdByRegion. See #HBASE-16994 for more details + wal.clearOldestUnflushedStoreSeqNum(encodedRegionName); + } else { + // Back up 1, minus 1 from oldest sequence id in memstore to get last 'flushed' edit + flushedSeqId = earliestUnflushedSequenceIdForTheRegion.longValue() - 1; + } } else { // use the provided sequence Id as WAL is not being used for this flush. flushedSeqId = flushOpSeqId = myseqid; @@ -2609,7 +2630,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi + prepareResult.totalFlushableSize.getDataSize() + ", currentsize=" + StringUtils.byteDesc(memstoresize) + "/" + memstoresize + " for region " + this + " in " + time + "ms, sequenceid=" - + flushOpSeqId + ", compaction requested=" + compactionRequested + + flushOpSeqId + ", maxFlushedSeqId=" + maxFlushedSeqId + + ", compaction requested=" + compactionRequested + ((wal == null) ? "; wal=null" : ""); LOG.info(msg); status.setStatus(msg); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java index 0ef0cf7..80241f5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java @@ -449,6 +449,11 @@ public abstract class AbstractFSWAL implements WAL { } @Override + public void clearOldestUnflushedStoreSeqNum(byte[] encodedRegionName) { + sequenceIdAccounting.clearOldestUnflushedStoreSeqNum(encodedRegionName); + } + + @Override public byte[][] rollWriter() throws FailedLogCloseException, IOException { return rollWriter(false); } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java index 62dea53..74bd094 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SequenceIdAccounting.java @@ -17,8 +17,6 @@ */ package org.apache.hadoop.hbase.regionserver.wal; -import com.google.common.annotations.VisibleForTesting; - import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -36,6 +34,8 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.ImmutableByteArray; +import com.google.common.annotations.VisibleForTesting; + /** * Accounting of sequence ids per region and then by column family. So we can our accounting * current, call startCacheFlush and then finishedCacheFlush or abortCacheFlush so this instance can @@ -325,6 +325,12 @@ class SequenceIdAccounting { } } + void clearOldestUnflushedStoreSeqNum(final byte[] encodedRegionName) { + synchronized (tieLock) { + this.lowestUnflushedSequenceIds.remove(encodedRegionName); + } + } + void abortCacheFlush(final byte[] encodedRegionName) { // Method is called when we are crashing down because failed write flush AND it is called // if we fail prepare. The below is for the fail prepare case; we restore the old sequence ids. diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/DisabledWALProvider.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/DisabledWALProvider.java index 0725c4e..9eb46ec 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/DisabledWALProvider.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/DisabledWALProvider.java @@ -30,10 +30,9 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; -import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.HRegionInfo; - +import org.apache.hadoop.hbase.classification.InterfaceAudience; // imports for things that haven't moved from regionserver.wal yet. import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener; import org.apache.hadoop.hbase.regionserver.wal.WALCoprocessorHost; @@ -232,6 +231,10 @@ class DisabledWALProvider implements WALProvider { @Override public void logRollerExited() { } + + @Override + public void clearOldestUnflushedStoreSeqNum(byte[] encodedRegionName) { + } } @Override diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java index c38c262..d4c4fe4 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WAL.java @@ -205,6 +205,12 @@ public interface WAL extends Closeable { long getEarliestMemstoreSeqNum(byte[] encodedRegionName, byte[] familyName); /** + * Clear oldestUnflushedStoreSequenceIdsOfRegion marker for all stores of the given region + * @param encodedRegionName Encoded region name. + */ + void clearOldestUnflushedStoreSeqNum(byte[] encodedRegionName); + + /** * Human readable identifying information about the state of this WAL. * Implementors are encouraged to include information appropriate for debugging. * Consumers are advised not to rely on the details of the returned String; it does