HBASE-20542-ADDENDUM: fix TestHStore
This commit is contained in:
parent
1e0650955a
commit
1804b6d059
|
@ -240,6 +240,15 @@ public class CompactingMemStore extends AbstractMemStore {
|
|||
return mss.getDataSize() > 0? mss: getActive().getMemStoreSize();
|
||||
}
|
||||
|
||||
|
||||
public void setInMemoryCompactionCompleted() {
|
||||
inMemoryCompactionInProgress.set(false);
|
||||
}
|
||||
|
||||
protected boolean setInMemoryCompactionFlag() {
|
||||
return inMemoryCompactionInProgress.compareAndSet(false, true);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected long keySize() {
|
||||
// Need to consider dataSize/keySize of all segments in pipeline and active
|
||||
|
@ -419,7 +428,7 @@ public class CompactingMemStore extends AbstractMemStore {
|
|||
if (shouldFlushInMemory(currActive, cellToAdd, memstoreSizing)) {
|
||||
if (currActive.setInMemoryFlushed()) {
|
||||
flushInMemory(currActive);
|
||||
if (inMemoryCompactionInProgress.compareAndSet(false, true)) {
|
||||
if (setInMemoryCompactionFlag()) {
|
||||
// The thread is dispatched to do in-memory compaction in the background
|
||||
InMemoryCompactionRunnable runnable = new InMemoryCompactionRunnable();
|
||||
if (LOG.isTraceEnabled()) {
|
||||
|
@ -455,21 +464,19 @@ public class CompactingMemStore extends AbstractMemStore {
|
|||
// setting the inMemoryCompactionInProgress flag again for the case this method is invoked
|
||||
// directly (only in tests) in the common path setting from true to true is idempotent
|
||||
inMemoryCompactionInProgress.set(true);
|
||||
// Used by tests
|
||||
if (!allowCompaction.get()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
// Used by tests
|
||||
if (!allowCompaction.get()) {
|
||||
return;
|
||||
// Speculative compaction execution, may be interrupted if flush is forced while
|
||||
// compaction is in progress
|
||||
if(!compactor.start()) {
|
||||
setInMemoryCompactionCompleted();
|
||||
}
|
||||
try {
|
||||
// Speculative compaction execution, may be interrupted if flush is forced while
|
||||
// compaction is in progress
|
||||
compactor.start();
|
||||
} catch (IOException e) {
|
||||
LOG.warn("Unable to run in-memory compaction on {}/{}; exception={}",
|
||||
getRegionServices().getRegionInfo().getEncodedName(), getFamilyName(), e);
|
||||
}
|
||||
} finally {
|
||||
inMemoryCompactionInProgress.set(false);
|
||||
} catch (IOException e) {
|
||||
LOG.warn("Unable to run in-memory compaction on {}/{}; exception={}",
|
||||
getRegionServices().getRegionInfo().getEncodedName(), getFamilyName(), e);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -147,20 +147,19 @@ public class MemStoreCompactor {
|
|||
private void doCompaction() {
|
||||
ImmutableSegment result = null;
|
||||
boolean resultSwapped = false;
|
||||
if (isInterrupted.get()) { // if the entire process is interrupted cancel flattening
|
||||
return; // the compaction also doesn't start when interrupted
|
||||
}
|
||||
|
||||
MemStoreCompactionStrategy.Action nextStep = strategy.getAction(versionedList);
|
||||
boolean merge =
|
||||
(nextStep == MemStoreCompactionStrategy.Action.MERGE ||
|
||||
nextStep == MemStoreCompactionStrategy.Action.MERGE_COUNT_UNIQUE_KEYS);
|
||||
boolean merge = (nextStep == MemStoreCompactionStrategy.Action.MERGE ||
|
||||
nextStep == MemStoreCompactionStrategy.Action.MERGE_COUNT_UNIQUE_KEYS);
|
||||
try {
|
||||
if (isInterrupted.get()) { // if the entire process is interrupted cancel flattening
|
||||
return; // the compaction also doesn't start when interrupted
|
||||
}
|
||||
|
||||
if (nextStep == MemStoreCompactionStrategy.Action.NOOP) {
|
||||
return;
|
||||
}
|
||||
if (nextStep == MemStoreCompactionStrategy.Action.FLATTEN ||
|
||||
nextStep == MemStoreCompactionStrategy.Action.FLATTEN_COUNT_UNIQUE_KEYS) {
|
||||
if (nextStep == MemStoreCompactionStrategy.Action.FLATTEN
|
||||
|| nextStep == MemStoreCompactionStrategy.Action.FLATTEN_COUNT_UNIQUE_KEYS) {
|
||||
// some Segment in the pipeline is with SkipList index, make it flat
|
||||
compactingMemStore.flattenOneSegment(versionedList.getVersion(), nextStep);
|
||||
return;
|
||||
|
@ -195,6 +194,7 @@ public class MemStoreCompactor {
|
|||
result.close();
|
||||
}
|
||||
releaseResources();
|
||||
compactingMemStore.setInMemoryCompactionCompleted();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -1738,15 +1738,15 @@ public class TestHStore {
|
|||
@Override
|
||||
public boolean start() throws IOException {
|
||||
boolean isFirst = RUNNER_COUNT.getAndIncrement() == 0;
|
||||
boolean rval = super.start();
|
||||
if (isFirst) {
|
||||
try {
|
||||
START_COMPACTOR_LATCH.await();
|
||||
return super.start();
|
||||
} catch (InterruptedException ex) {
|
||||
throw new RuntimeException(ex);
|
||||
}
|
||||
}
|
||||
return rval;
|
||||
return super.start();
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -1765,12 +1765,15 @@ public class TestHStore {
|
|||
}
|
||||
|
||||
@Override
|
||||
void inMemoryCompaction() {
|
||||
RUNNER_COUNT.incrementAndGet();
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("runner count: " + RUNNER_COUNT.get());
|
||||
protected boolean setInMemoryCompactionFlag() {
|
||||
boolean rval = super.setInMemoryCompactionFlag();
|
||||
if (rval) {
|
||||
RUNNER_COUNT.incrementAndGet();
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("runner count: " + RUNNER_COUNT.get());
|
||||
}
|
||||
}
|
||||
super.inMemoryCompaction();
|
||||
return rval;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue