org.apache.commons.lang3.mutable.MutableBoolean#setValue ( )源码实例Demo

/**
 * Used to fetch the first data holding batch from either the build or probe side.
 * @param outcome The current upstream outcome for either the build or probe side.
 * @param prefetched A flag indicating if we have already done a prefetch of the first data holding batch for the probe or build side.
 * @param isEmpty A flag indicating if the probe or build side is empty.
 * @param index The upstream index of the probe or build batch.
 * @param batch The probe or build batch itself.
 * @param memoryManagerUpdate A lambda function to execute the memory manager update for the probe or build batch.
 * @return The current {@link org.apache.drill.exec.record.RecordBatch.IterOutcome}.
 */
private IterOutcome prefetchFirstBatch(IterOutcome outcome,
                                       final MutableBoolean prefetched,
                                       final MutableBoolean isEmpty,
                                       final int index,
                                       final RecordBatch batch,
                                       final Runnable memoryManagerUpdate) {
  if (prefetched.booleanValue()) {
    // We have already prefetch the first data holding batch
    return outcome;
  }

  // If we didn't retrieve our first data holding batch, we need to do it now.
  prefetched.setValue(true);

  if (outcome != IterOutcome.NONE) {
    // We can only get data if there is data available
    outcome = sniffNonEmptyBatch(outcome, index, batch);
  }

  isEmpty.setValue(outcome == IterOutcome.NONE); // If we recieved NONE there is no data.

  if (outcome == IterOutcome.OUT_OF_MEMORY) {
    // We reached a termination state
    state = BatchState.OUT_OF_MEMORY;
  } else if (outcome == IterOutcome.STOP) {
    // We reached a termination state
    state = BatchState.STOP;
  } else {
    // Got our first batch(es)
    if (spilledState.isFirstCycle()) {
      // Only collect stats for the first cylce
      memoryManagerUpdate.run();
    }
    state = BatchState.FIRST;
  }

  return outcome;
}
 

private boolean isExpectedElementsQuantity(SearchAttributes locator, Matcher<Integer> elementsMatcher,
        MutableBoolean firstIteration)
{
    if (firstIteration.isTrue())
    {
        firstIteration.setValue(false);
        return softAssert.assertThat("Elements number", getElementsNumber(locator), elementsMatcher);
    }
    return elementsMatcher.matches(getElementsNumber(locator));
}
 

private static String truncate(String string, int maxLen, MutableBoolean truncated) {
    if (string != null && string.length() > maxLen) {
        truncated.setValue(true);
        return string.substring(0, maxLen - 3) + "...";
    }
    return string;
}
 

@Test
public void expiredTestBlocking() throws Exception
{
  SimpleDoneQueueManager<Query, Void> sdqqm = new SimpleDoneQueueManager<Query, Void>();

  sdqqm.setup(null);
  sdqqm.beginWindow(0);

  Query query = new MockQuery("1");
  MutableBoolean queueContext = new MutableBoolean(false);
  sdqqm.enqueue(query, null, queueContext);

  Assert.assertEquals(1, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());
  QueryBundle<Query, Void, MutableBoolean> qb = sdqqm.dequeueBlock();
  Assert.assertEquals(0, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  sdqqm.endWindow();

  sdqqm.beginWindow(1);

  Assert.assertEquals(1, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  queueContext.setValue(true);
  testBlocking(sdqqm);

  Assert.assertEquals(0, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  sdqqm.endWindow();
  sdqqm.teardown();
}
 

@Test
public void simpleExpireBlockThenUnblock() throws Exception
{
  SimpleDoneQueueManager<Query, Void> sdqqm = new SimpleDoneQueueManager<Query, Void>();

  sdqqm.setup(null);
  sdqqm.beginWindow(0);

  Query query = new MockQuery("1");
  MutableBoolean expire = new MutableBoolean(false);
  sdqqm.enqueue(query, null, expire);

  sdqqm.endWindow();
  sdqqm.beginWindow(1);

  //Expire
  expire.setValue(true);

  ExceptionSaverExceptionHandler eseh = new ExceptionSaverExceptionHandler();
  testBlockingNoStop(sdqqm, eseh);

  query = new MockQuery("2");
  sdqqm.enqueue(query, null, new MutableBoolean(false));

  Thread.sleep(1000);

  Assert.assertNull(eseh.getCaughtThrowable());

  sdqqm.endWindow();
  sdqqm.teardown();
}
 

public static <T> Iterable<T> iterable(Iterator<T> seq) {
    final MutableBoolean used = new MutableBoolean(false);
    return new Iterable<T>() {

        @Override
        public Iterator<T> iterator() {
            if (!used.booleanValue()) {
                used.setValue(true);
                return seq;
            } else {
                throw new IllegalStateException("only allowed to iterate this iterable once");
            }
        }
    };
}
 

@Test
public void test() throws IOException, InterruptedException {
  final HRegionServer rs = testUtil.getRSForFirstRegionInTable(tableName);
  final HRegion region = (HRegion) rs.getRegions(tableName).get(0);
  HRegion spiedRegion = spy(region);
  final MutableBoolean flushed = new MutableBoolean(false);
  final MutableBoolean reported = new MutableBoolean(false);
  doAnswer(new Answer<FlushResult>() {
    @Override
    public FlushResult answer(InvocationOnMock invocation) throws Throwable {
      synchronized (flushed) {
        flushed.setValue(true);
        flushed.notifyAll();
      }
      synchronized (reported) {
        while (!reported.booleanValue()) {
          reported.wait();
        }
      }
      rs.getWAL(region.getRegionInfo()).abortCacheFlush(
        region.getRegionInfo().getEncodedNameAsBytes());
      throw new DroppedSnapshotException("testcase");
    }
  }).when(spiedRegion).internalFlushCacheAndCommit(Matchers.<WAL> any(),
    Matchers.<MonitoredTask> any(), Matchers.<PrepareFlushResult> any(),
    Matchers.<Collection<HStore>> any());
  // Find region key; don't pick up key for hbase:meta by mistake.
  String key = null;
  for (Map.Entry<String, HRegion> entry: rs.getOnlineRegions().entrySet()) {
    if (entry.getValue().getRegionInfo().getTable().equals(this.tableName)) {
      key = entry.getKey();
      break;
    }
  }
  rs.getOnlineRegions().put(key, spiedRegion);
  Connection conn = testUtil.getConnection();

  try (Table table = conn.getTable(tableName)) {
    table.put(new Put(Bytes.toBytes("row0"))
            .addColumn(family, qualifier, Bytes.toBytes("val0")));
  }
  long oldestSeqIdOfStore = region.getOldestSeqIdOfStore(family);
  LOG.info("CHANGE OLDEST " + oldestSeqIdOfStore);
  assertTrue(oldestSeqIdOfStore > HConstants.NO_SEQNUM);
  rs.getMemStoreFlusher().requestFlush(spiedRegion, FlushLifeCycleTracker.DUMMY);
  synchronized (flushed) {
    while (!flushed.booleanValue()) {
      flushed.wait();
    }
  }
  try (Table table = conn.getTable(tableName)) {
    table.put(new Put(Bytes.toBytes("row1"))
            .addColumn(family, qualifier, Bytes.toBytes("val1")));
  }
  long now = EnvironmentEdgeManager.currentTime();
  rs.tryRegionServerReport(now - 500, now);
  synchronized (reported) {
    reported.setValue(true);
    reported.notifyAll();
  }
  while (testUtil.getRSForFirstRegionInTable(tableName) == rs) {
    Thread.sleep(100);
  }
  try (Table table = conn.getTable(tableName)) {
    Result result = table.get(new Get(Bytes.toBytes("row0")));
    assertArrayEquals(Bytes.toBytes("val0"), result.getValue(family, qualifier));
  }
}
 

@Test
public void expiredTestBlockingExpiredFirstValidLast() throws Exception
{
  SimpleDoneQueueManager<Query, Void> sdqqm = new SimpleDoneQueueManager<Query, Void>();

  sdqqm.setup(null);
  sdqqm.beginWindow(0);

  Query query = new MockQuery("1");
  MutableBoolean queueContext = new MutableBoolean(false);
  sdqqm.enqueue(query, null, queueContext);

  Query query1 = new MockQuery("2");
  MutableBoolean queueContext1 = new MutableBoolean(false);
  sdqqm.enqueue(query1, null, queueContext1);

  Assert.assertEquals(2, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());
  QueryBundle<Query, Void, MutableBoolean> qb = sdqqm.dequeueBlock();
  Assert.assertEquals(1, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  sdqqm.endWindow();

  sdqqm.beginWindow(1);

  Assert.assertEquals(2, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  queueContext.setValue(true);
  qb = sdqqm.dequeueBlock();

  Assert.assertEquals(0, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  testBlocking(sdqqm);

  sdqqm.endWindow();

  sdqqm.beginWindow(2);

  Assert.assertEquals(1, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  qb = sdqqm.dequeueBlock();

  testBlocking(sdqqm);

  sdqqm.endWindow();

  sdqqm.teardown();
}
 

@Test
public void expiredTestBlockingValidFirstExpiredLast() throws Exception
{
  SimpleDoneQueueManager<Query, Void> sdqqm = new SimpleDoneQueueManager<Query, Void>();

  sdqqm.setup(null);
  sdqqm.beginWindow(0);

  Query query = new MockQuery("1");
  MutableBoolean queueContext = new MutableBoolean(false);
  sdqqm.enqueue(query, null, queueContext);

  Query query1 = new MockQuery("2");
  MutableBoolean queueContext1 = new MutableBoolean(false);
  sdqqm.enqueue(query1, null, queueContext1);

  Assert.assertEquals(2, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());
  QueryBundle<Query, Void, MutableBoolean> qb = sdqqm.dequeueBlock();
  Assert.assertEquals(1, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  sdqqm.endWindow();

  sdqqm.beginWindow(1);

  Assert.assertEquals(2, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  queueContext1.setValue(true);
  qb = sdqqm.dequeueBlock();

  Assert.assertEquals(1, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  testBlocking(sdqqm);

  Assert.assertEquals(0, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  sdqqm.endWindow();

  sdqqm.beginWindow(2);

  Assert.assertEquals(1, sdqqm.getNumLeft());
  Assert.assertEquals(sdqqm.getNumPermits(), sdqqm.getNumLeft());

  qb = sdqqm.dequeueBlock();

  testBlocking(sdqqm);

  sdqqm.endWindow();

  sdqqm.teardown();
}