/**
 * Test case for Bug #1905.
 *
 * @throws Exception if the test fails
 */
@Test
public void getChildElements() throws Exception {
    final String xml = "<events>\n"
            + "  <something/>\n"
            + "</events>";
    getMockWebConnection().setDefaultResponse(xml, MimeType.TEXT_XML);
    getWebClient().setWebConnection(getMockWebConnection());
    final XmlPage page = getWebClient().getPage(URL_FIRST);
    final DomElement root = page.getDocumentElement();
    final AtomicInteger count = new AtomicInteger(0);
    root.getChildElements().forEach(e -> count.incrementAndGet());
    assertEquals(1, count.get());

    count.set(0);
    root.getChildren().forEach(e -> count.incrementAndGet());
    assertEquals(3, count.get());
}
 

/**
 * 1) check the regions is allowed. 2) check the concurrent tasks for
 * regions. 3) check the total concurrent tasks. 4) check the concurrent
 * tasks for server.
 *
 * @param loc the destination of data
 * @param heapSizeOfRow the data size
 * @return either Include {@link RequestController.ReturnCode} or skip
 *         {@link RequestController.ReturnCode}
 */
@Override
public ReturnCode canTakeOperation(HRegionLocation loc, long heapSizeOfRow) {
  RegionInfo regionInfo = loc.getRegion();
  if (regionsIncluded.contains(regionInfo)) {
    // We already know what to do with this region.
    return ReturnCode.INCLUDE;
  }
  AtomicInteger regionCnt = taskCounterPerRegion.get(loc.getRegion().getRegionName());
  if (regionCnt != null && regionCnt.get() >= maxConcurrentTasksPerRegion) {
    // Too many tasks on this region already.
    return ReturnCode.SKIP;
  }
  int newServers = serversIncluded.size()
          + (serversIncluded.contains(loc.getServerName()) ? 0 : 1);
  if ((newServers + tasksInProgress.get()) > maxTotalConcurrentTasks) {
    // Too many tasks.
    return ReturnCode.SKIP;
  }
  AtomicInteger serverCnt = taskCounterPerServer.get(loc.getServerName());
  if (serverCnt != null && serverCnt.get() >= maxConcurrentTasksPerServer) {
    // Too many tasks for this individual server
    return ReturnCode.SKIP;
  }
  return ReturnCode.INCLUDE;
}
 

@Test
public void runTest() {
  AtomicInteger count = new AtomicInteger();
  AtomicBoolean sameContext = new AtomicBoolean();
  TestSuite suite = TestSuite.create("my_suite").
      test("my_test", context -> {
        sameContext.set(checkTest(context));
        count.compareAndSet(0, 1);
      });
  TestReporter reporter = new TestReporter();
  run(suite, reporter);
  reporter.await();
  assertTrue(sameContext.get());
  assertEquals(1, count.get());
  assertEquals(0, reporter.exceptions.size());
  assertEquals(1, reporter.results.size());
  TestResult result = reporter.results.get(0);
  assertEquals("my_test", result.name());
  assertTrue(result.succeeded());
  assertFalse(result.failed());
  assertNull(result.failure());
}
 

public TimedHandler(MeterRegistry registry, Iterable<Tag> tags, HttpServletRequestTagsProvider tagsProvider) {
    this.registry = registry;
    this.tags = tags;
    this.tagsProvider = tagsProvider;

    this.openRequests = LongTaskTimer.builder("jetty.server.dispatches.open")
            .description("Jetty dispatches that are currently in progress")
            .tags(tags)
            .register(registry);

    this.asyncDispatches = Counter.builder("jetty.server.async.dispatches")
            .description("Asynchronous dispatches")
            .tags(tags)
            .register(registry);

    this.asyncExpires = Counter.builder("jetty.server.async.expires")
            .description("Asynchronous operations that timed out before completing")
            .tags(tags)
            .register(registry);

    Gauge.builder("jetty.server.async.waits", asyncWaits, AtomicInteger::doubleValue)
            .description("Pending asynchronous wait operations")
            .baseUnit(BaseUnits.OPERATIONS)
            .tags(tags)
            .register(registry);
}
 

/**
 * Formats a number and appends the resulting text to the given string
 * buffer.
 * The number can be of any subclass of {@link java.lang.Number}.
 * <p>
 * This implementation uses the maximum precision permitted.
 * @param number     the number to format
 * @param toAppendTo the <code>StringBuffer</code> to which the formatted
 *                   text is to be appended
 * @param pos        On input: an alignment field, if desired.
 *                   On output: the offsets of the alignment field.
 * @return           the value passed in as <code>toAppendTo</code>
 * @exception        IllegalArgumentException if <code>number</code> is
 *                   null or not an instance of <code>Number</code>.
 * @exception        NullPointerException if <code>toAppendTo</code> or
 *                   <code>pos</code> is null
 * @exception        ArithmeticException if rounding is needed with rounding
 *                   mode being set to RoundingMode.UNNECESSARY
 * @see              java.text.FieldPosition
 */
@Override
public final StringBuffer format(Object number,
                                 StringBuffer toAppendTo,
                                 FieldPosition pos) {
    if (number instanceof Long || number instanceof Integer ||
               number instanceof Short || number instanceof Byte ||
               number instanceof AtomicInteger ||
               number instanceof AtomicLong ||
               (number instanceof BigInteger &&
                ((BigInteger)number).bitLength () < 64)) {
        return format(((Number)number).longValue(), toAppendTo, pos);
    } else if (number instanceof BigDecimal) {
        return format((BigDecimal)number, toAppendTo, pos);
    } else if (number instanceof BigInteger) {
        return format((BigInteger)number, toAppendTo, pos);
    } else if (number instanceof Number) {
        return format(((Number)number).doubleValue(), toAppendTo, pos);
    } else {
        throw new IllegalArgumentException("Cannot format given Object as a Number");
    }
}
 

@Test
public void remove_existingPeer() {
  // Initial add
  final Peer peer = createPeer();
  assertThat(maintainedPeers.add(peer)).isTrue();
  assertThat(maintainedPeers.size()).isEqualTo(1);
  assertThat(maintainedPeers.contains(peer)).isTrue();

  // Test remove
  final AtomicInteger callbackCount = new AtomicInteger(0);
  maintainedPeers.subscribeRemove(
      (addedPeer, wasRemoved) -> {
        callbackCount.incrementAndGet();
        assertThat(addedPeer).isEqualTo(peer);
        assertThat(wasRemoved).isTrue();
      });
  assertThat(maintainedPeers.remove(peer)).isTrue();
  assertThat(callbackCount).hasValue(1);
  assertThat(maintainedPeers.size()).isEqualTo(0);
  assertThat(maintainedPeers.contains(peer)).isFalse();
}
 

@Test
public void testExceptionInSubscriberTerminatesSubscription() {
    AtomicInteger errorCounter = new AtomicInteger();
    Disposable subscription = combined.subscribe(
            next -> {
                if (next.equals("T3")) {
                    throw new RuntimeException("simulated error");
                }
            },
            e -> errorCounter.incrementAndGet()
    );

    // Wait until hot observable emits error
    while (!terminateFlag.get()) {
    }

    await().timeout(5, TimeUnit.SECONDS).until(subscription::isDisposed);
    assertThat(errorCounter.get()).isEqualTo(1);
}
 

void test(String[] args) throws Throwable {
    for (int how=0; how<4; how++) {
        final CountDownLatch done = new CountDownLatch(1);
        final AtomicInteger count = new AtomicInteger(0);
        final Timer timer = new Timer();
        final TimerTask task = new TimerTask() {
            @Override
            public void run() {
                checkScheduledExecutionTime(this);
                count.incrementAndGet();
                done.countDown();
            }};

        scheduleNow(timer, task, how);
        done.await();
        equal(count.get(), 1);
        checkScheduledExecutionTime(task);
        if (new java.util.Random().nextBoolean())
            sleep(10);
        check(task.cancel());
        timer.cancel();
        checkScheduledExecutionTime(task);
    }
}
 

public CompletableFuture<List<DMLResult>> executeUpdatesAsync(
        String tableSpace, String query, long tx, boolean returnValues,
        boolean usePreparedStatement, List<List<Object>> batch
) {
    if (batch.isEmpty()) {
        return CompletableFuture.completedFuture(Collections.emptyList());
    }
    if (discoverTablespaceFromSql) {
        tableSpace = discoverTablespace(tableSpace, query);
    }
    if (closed) {
        return FutureUtils.exception(new HDBException("client is closed"));
    }
    CompletableFuture<List<DMLResult>> res = new CompletableFuture<>();

    AtomicInteger count = new AtomicInteger(0);
    executeStatementsAsyncInternal(tableSpace, res, query, tx, returnValues, usePreparedStatement, batch, count);
    return res;
}
 

@Override
public boolean execute(Object localId, Object remoteId, Object... args) {
  if (localId == null) {
    return false;
  }
  final String localIdStr = localId.toString();
  final AtomicInteger d = delays.get(localIdStr);
  if (d == null) {
    return false;
  }
  LOG.info("{} delay {} ms, args={}", localIdStr, d.get(),
      Arrays.toString(args));
  try {
    RaftTestUtil.delay(d::get);
  } catch (InterruptedException e) {
    LOG.debug("Interrupted while delaying " + localIdStr);
  }
  return true;
}
 

@Test
public void echoServiceUsingPayloadWriterWithSerializerWithTrailers() throws Exception {
    echoService((ctx, request, response) -> {
        response.setHeader(TRAILER, X_TOTAL_LENGTH);
        try (HttpPayloadWriter<String> pw = response.sendMetaData(textSerializer())) {
            AtomicInteger totalLength = new AtomicInteger();
            request.payloadBody(textDeserializer()).forEach(chunk -> {
                try {
                    totalLength.addAndGet(chunk.length());
                    pw.write(chunk);
                } catch (IOException e) {
                    throwException(e);
                }
            });
            pw.setTrailer(X_TOTAL_LENGTH, totalLength.toString());
        }
    });
}
 

@SuppressWarnings("unchecked")
public KotlinRepl(KotlinReplProperties properties) {
  compiler = ReplBuilding.buildCompiler(properties);
  evaluator = ReplBuilding.buildEvaluator(properties);
  ReentrantReadWriteLock stateLock = new ReentrantReadWriteLock();
  state = new AggregatedReplStageState(
      compiler.createState(stateLock),
      evaluator.createState(stateLock),
      stateLock);
  counter = new AtomicInteger(0);

  writer = new ClassWriter(properties.getOutputDir());
  maxResult = properties.getMaxResult();
  shortenTypes = properties.getShortenTypes();

  ctx = new KotlinContext();
  properties.getReceiver().kc = ctx;

  contextUpdater = new ContextUpdater(
      state, ctx.vars, ctx.functions);

  for (String line: properties.getCodeOnLoad()) {
    eval(line);
  }
}
 

@Test
public void testIterateForwards() {
    final RingBuffer<Integer> ringBuffer = new RingBuffer<>(10);

    final int[] values = new int[]{3, 5, 20, 7};
    for (final int v : values) {
        ringBuffer.add(v);
    }

    final AtomicInteger countHolder = new AtomicInteger(0);
    ringBuffer.forEach(new ForEachEvaluator<Integer>() {
        int counter = 0;

        @Override
        public boolean evaluate(final Integer value) {
            final int expected = values[counter++];
            countHolder.incrementAndGet();
            assertEquals(expected, value.intValue());
            return true;
        }

    }, IterationDirection.FORWARD);

    assertEquals(4, countHolder.get());
}
 

private void verify(String clazz, byte[] data, Map<String,Boolean> loadable, ClassLoader loader, AtomicInteger maxWarn)
        throws IOException, BuildException {
    //log("Verifying linkage of " + clazz.replace('/', '.'), Project.MSG_DEBUG);
    Set<String> dependencies = dependencies(data);
    //System.err.println(clazz + " -> " + dependencies);
    for (String clazz2 : dependencies) {
        Boolean exists = loadable.get(clazz2);
        if (exists == null) {
            exists = loader.getResource(clazz2.replace('.', '/') + ".class") != null;
            loadable.put(clazz2, exists);
        }
        if (!exists) {
            String message = clazz + " cannot access " + clazz2;
            if (failOnError) {
                throw new BuildException(message, getLocation());
            } else if (maxWarn.getAndDecrement() > 0) {
                log("Warning: " + message, Project.MSG_WARN);
            } else {
                log("(additional warnings not reported)", Project.MSG_WARN);
                return;
            }
        } else {
            //log("Working reference to " + clazz2, Project.MSG_DEBUG);
        }
    }
}
 

public WeightedRoundRobinMultiplexer(int aNumQueues, String ns,
  Configuration conf) {
  if (aNumQueues <= 0) {
    throw new IllegalArgumentException("Requested queues (" + aNumQueues +
      ") must be greater than zero.");
  }

  this.numQueues = aNumQueues;
  this.queueWeights = conf.getInts(ns + "." +
    IPC_CALLQUEUE_WRRMUX_WEIGHTS_KEY);

  if (this.queueWeights.length == 0) {
    this.queueWeights = getDefaultQueueWeights(this.numQueues);
  } else if (this.queueWeights.length != this.numQueues) {
    throw new IllegalArgumentException(ns + "." +
      IPC_CALLQUEUE_WRRMUX_WEIGHTS_KEY + " must specify exactly " +
      this.numQueues + " weights: one for each priority level.");
  }

  this.currentQueueIndex = new AtomicInteger(0);
  this.requestsLeft = new AtomicInteger(this.queueWeights[0]);

  LOG.info("WeightedRoundRobinMultiplexer is being used.");
}
 

private void setFieldsByConfiguration(final Configuration config, final boolean update) {
	if (!update || this.isPropertyUpdateable(CONFIG_PROPERTY_NAME_DELTA_TIME)) {
		this.deltat = new AtomicLong(config.getLongProperty(CONFIG_PROPERTY_NAME_DELTA_TIME));
	}

	if (!update || this.isPropertyUpdateable(CONFIG_PROPERTY_NAME_DELTA_UNIT)) {
		this.tunit = TimeUnit.valueOf(config.getStringProperty(CONFIG_PROPERTY_NAME_DELTA_UNIT));
	}

	if (!update || this.isPropertyUpdateable(CONFIG_PROPERTY_NAME_FC_METHOD)) {
		this.forecastMethod.set(ForecastMethod.valueOf(config.getStringProperty(CONFIG_PROPERTY_NAME_FC_METHOD)));
	}

	if (!update || this.isPropertyUpdateable(CONFIG_PROPERTY_NAME_TS_WINDOW_CAPACITY)) {
		this.timeSeriesWindowCapacity = new AtomicInteger(config.getIntProperty(CONFIG_PROPERTY_NAME_TS_WINDOW_CAPACITY));
	}

	if (!update || this.isPropertyUpdateable(CONFIG_PROPERTY_NAME_FC_CONFIDENCE)) {
		this.forecastConfidence = new AtomicInteger(config.getIntProperty(CONFIG_PROPERTY_NAME_FC_CONFIDENCE));
	}
}
 

@Test
public void test_received_request_twoway_error_reqeustBroken() throws RemotingException{
    final Request request = new Request();
    request.setTwoWay(true);
    request.setData(new BizException());
    request.setBroken(true);
    
    final AtomicInteger count = new AtomicInteger(0);
    final Channel mchannel = new MockedChannel(){
        @Override
        public void send(Object message) throws RemotingException {
            Response res = (Response)message;
            Assert.assertEquals(request.getId(), res.getId());
            Assert.assertEquals(request.getVersion(), res.getVersion());
            Assert.assertEquals(Response.BAD_REQUEST, res.getStatus());
            Assert.assertNull(res.getResult());
            Assert.assertTrue(res.getErrorMessage().contains(BizException.class.getName()));
            count.incrementAndGet();
        }
    };
    HeaderExchangeHandler hexhandler = new HeaderExchangeHandler(new MockedExchangeHandler());
    hexhandler.received(mchannel, request);
    Assert.assertEquals(1, count.get());
}
 

@Test
public void attachEvent_stateTreeCanFound() {
    ShadowRoot bodyShadow = new UI().getElement().attachShadow();
    Element child = ElementFactory.createDiv();

    AtomicInteger attached = new AtomicInteger();

    child.addAttachListener(event -> {
        Assert.assertNotNull(event.getSource().getNode().getOwner());
        Assert.assertNotEquals(NullOwner.get(),
                event.getSource().getNode().getOwner());
    });
    child.addAttachListener(event -> attached.incrementAndGet());

    bodyShadow.appendChild(child);
    Assert.assertEquals(1, attached.get());
}
 

@Test
public void onlyAccessSpiPackagesTest()
        throws ClassNotFoundException, IOException
{
    Module module = pluginLoader.getModules().get(0);
    ClassLoader moduleClassLoader = module.getModuleClassLoader();

    try {
        moduleClassLoader.loadClass(PluginLoader.class.getName());
        Assert.fail();
    }
    catch (ClassNotFoundException ignored) {
    }

    moduleClassLoader.loadClass(AopFactory.class.getName());
    moduleClassLoader.loadClass(AtomicInteger.class.getName());
    Assert.assertTrue(Driver.class.isAssignableFrom(moduleClassLoader.loadClass("org.h2.Driver")));

    Assert.assertNull(moduleClassLoader.getResource("version1"));

    Assert.assertNotNull(moduleClassLoader.getResources("version2").nextElement());
    Assert.assertNotNull(moduleClassLoader.getResource("version2"));

    Assert.assertNotNull(this.getClass().getClassLoader().getResource("version1"));
}
 

/**
 * @throws Exception If failed.
 */
@Params(baseline = 9, atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
@Test
public void testStopBaselineTransactionalReplicated() throws Exception {
    AtomicInteger cntDownCntr = new AtomicInteger(0);

    doTest(
        asMessagePredicate(discoEvt -> discoEvt.type() == EventType.EVT_NODE_LEFT),
        () -> {
            IgniteEx node = baseline.get(baseline.size() - cntDownCntr.get() - 1);

            TestRecordingCommunicationSpi.spi(node).stopBlock();

            cntDownCntr.incrementAndGet();

            for (int i = 0; i < cntDownCntr.get(); i++)
                cntFinishedReadOperations.countDown(); // This node and previously stopped nodes as well.

            stopGrid(node.name());
        }
    );
}
 

public ZkUtils(ZkKeyBuilder zkKeyBuilder, ZkClient zkClient, int connectionTimeoutMs, int sessionTimeOutMs, MetricsRegistry metricsRegistry) {
  this.keyBuilder = zkKeyBuilder;
  this.connectionTimeoutMs = connectionTimeoutMs;
  this.zkClient = zkClient;
  this.metrics = new ZkUtilsMetrics(metricsRegistry);
  this.currentGeneration = new AtomicInteger(0);
  this.sessionTimeoutMs = sessionTimeOutMs;
}
 

@Test
public void testRunningJobsAreInterruptedAfterShutdownNow() throws InterruptedException {
  final int jobCount = 4;

  BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
  ExecutorService service = Executors.newCachedThreadPool();
  try {
    PartitionedUnorderedExecutor executor = new PartitionedUnorderedExecutor(queue, service, jobCount);

    final Semaphore jobSemaphore = new Semaphore(0);
    final Semaphore testSemaphore = new Semaphore(0);
    final AtomicInteger interrupted = new AtomicInteger();

    for (int i = 0; i < jobCount; i++) {
      executor.submit(() -> {
        testSemaphore.release();
        try {
          jobSemaphore.acquire();
        } catch (InterruptedException e) {
          interrupted.incrementAndGet();
        }
      });
    }

    testSemaphore.acquireUninterruptibly(jobCount);

    assertThat(executor.shutdownNow(), empty());
    assertThat(executor.awaitTermination(2, MINUTES), is(true));
    assertThat(executor.isShutdown(), is(true));
    assertThat(executor.isTerminated(), is(true));

    assertThat(jobSemaphore.availablePermits(), is(0));
    assertThat(interrupted.get(), is(jobCount));
  } finally {
    service.shutdown();
  }
}
 

@Override
protected void moveToSecondPhase() {
    final AggregatedDfs dfs = searchPhaseController.aggregateDfs(firstResults);
    final AtomicInteger counter = new AtomicInteger(firstResults.asList().size());

    for (final AtomicArray.Entry<DfsSearchResult> entry : firstResults.asList()) {
        DfsSearchResult dfsResult = entry.value;
        DiscoveryNode node = nodes.get(dfsResult.shardTarget().nodeId());
        QuerySearchRequest querySearchRequest = new QuerySearchRequest(request, dfsResult.id(), dfs);
        executeSecondPhase(entry.index, dfsResult, counter, node, querySearchRequest);
    }
}
 

@Test
public void testRejectException() throws InterruptedException {
    final StringBuilder otherException = new StringBuilder();
    final AtomicInteger count = new AtomicInteger();
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(16);
    for(int i = 0; i < 16; i++) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 4; i++) {
                    try {
                        String result = client.testString("string");
                        System.out.println("testString: " + result);
                    } catch (Exception e) {
                        if (e.getCause().getMessage().contains("RejectedExecutionException")) {
                            count.incrementAndGet();
                        } else {
                            otherException.append(e.getCause().getClass().getName());
                        }
                    }
                }
            }
        });
    }
    executorService.shutdown();
    while(true){
        if(executorService.isTerminated()){
            System.out.println("所有的子线程都结束了！");
            break;
        }
        Thread.sleep(1000);
    }
    Assert.assertEquals("", otherException.toString());
    Assert.assertTrue(count.get() > 0);
}
 

@Test
public void monoCacheContextHistory() {
	AtomicInteger contextFillCount = new AtomicInteger();
	Mono<String> cached = Mono.subscriberContext()
	                          .map(ctx -> ctx.getOrDefault("a", "BAD"))
	                          .cache()
	                          .subscriberContext(ctx -> ctx.put("a", "GOOD" + contextFillCount.incrementAndGet()));

	//at first pass, the context is captured
	String cacheMiss = cached.block();
	Assertions.assertThat(cacheMiss).as("cacheMiss").isEqualTo("GOOD1");
	Assertions.assertThat(contextFillCount).as("cacheMiss").hasValue(1);

	//at second subscribe, the Context fill attempt is still done, but ultimately ignored since first context is cached
	String cacheHit = cached.block();
	Assertions.assertThat(cacheHit).as("cacheHit").isEqualTo("GOOD1"); //value from the cache
	Assertions.assertThat(contextFillCount).as("cacheHit").hasValue(2); //function was still invoked

	//at third subscribe, function is called for the 3rd time, but the context is still cached
	String cacheHit2 = cached.block();
	Assertions.assertThat(cacheHit2).as("cacheHit2").isEqualTo("GOOD1");
	Assertions.assertThat(contextFillCount).as("cacheHit2").hasValue(3);

	//at fourth subscribe, function is called for the 4th time, but the context is still cached
	String cacheHit3 = cached.block();
	Assertions.assertThat(cacheHit3).as("cacheHit3").isEqualTo("GOOD1");
	Assertions.assertThat(contextFillCount).as("cacheHit3").hasValue(4);
}
 

public StartEnabledAddonCallable(Furnace furnace,
         AddonLifecycleManager lifecycleManager,
         AddonStateManager stateManager,
         ExecutorService executor,
         AtomicInteger starting,
         Addon toStart)
{
   this.furnace = furnace;
   this.lifecycleManager = lifecycleManager;
   this.stateManager = stateManager;
   this.executor = executor;
   this.starting = starting;
   this.addon = toStart;
}
 

/**
 * Create a topology for the publisher application and run it.
 */
private void run() throws Exception {
    DevelopmentProvider tp = new DevelopmentProvider();
    
    // build the application/topology
    
    Topology t = tp.newTopology("mqttSamplePublisher");

    // Create the MQTT broker connector
    MqttConfig mqttConfig = createMqttConfig();
    MqttStreams mqtt = new MqttStreams(t, () -> mqttConfig);
    
    // Create a sample stream of tuples to publish
    AtomicInteger cnt = new AtomicInteger();
    TStream<String> msgs = t.poll(
            () -> {
                String msg = String.format("Message-%d from %s",
                        cnt.incrementAndGet(), Util.simpleTS());
                System.out.println("poll generated msg to publish: " + msg);
                return msg;
            }, 1L, TimeUnit.SECONDS);
    
    // Publish the stream to the topic.  The String tuple is the message value.
    mqtt.publish(msgs, topic, 0/*qos*/, false/*retain*/);
    
    // run the application / topology
    System.out.println("Console URL for the job: "
            + tp.getServices().getService(HttpServer.class).getConsoleUrl());
    tp.submit(t);
}
 

public int getEventCount(Events type) {
    int retVal = 0;
    lock.lock();
    try {
        AtomicInteger counter = eventCounts.get(type);
        if (counter != null) {
            retVal = counter.get();
        }
    } finally {
        lock.unlock();
    }
    return retVal;
}
 

@Test
public void testFuture() {

    SettableFuture<Boolean> objectSettableFuture = SettableFuture.create();
    AtomicInteger listenerCount = new AtomicInteger(0);
    AtomicInteger notifyCount = new AtomicInteger();

    executors.execute(new AbstractExceptionLogTask() {

        @Override
        protected void doRun() throws Exception {

            while (!Thread.interrupted()) {

                listenerCount.incrementAndGet();
                objectSettableFuture.addListener(new Runnable() {
                    @Override
                    public void run() {
                        notifyCount.incrementAndGet();
                    }
                }, MoreExecutors.directExecutor());
            }

            logger.info("exit thread");
        }
    });

    sleep(10);
    objectSettableFuture.set(true);

    executors.shutdownNow();
    sleep(10);

    logger.info("{}, {}", listenerCount, notifyCount);
    Assert.assertEquals(listenerCount.get(), notifyCount.get());
}
 

/**
 * Decreases in-use count for an orphan. If count reaches 0, the orphan is removed.
 */
private synchronized boolean decreaseOrphansUsageCountAndMaybeRemove(
    final CacheEntry<K, V> cacheEntry) {
  AtomicInteger counter = mOrphans.get(cacheEntry);
  Preconditions.checkNotNull(counter);
  Preconditions.checkState(counter.get() > 0);
  if (counter.decrementAndGet() == 0) {
    mOrphans.remove(cacheEntry);
    return true;
  }
  return false;
}