private void concurrentOperationsDunitTest(final boolean createBlockingQueue,
      final Scope rscope) {
    // Create Cache and HARegionQueue in all the 4 VMs.

    CacheSerializableRunnable createRgnsAndQueues = new CacheSerializableRunnable(
        "CreateCache, mirrored Region & HARegionQueue with a CacheListener") {
      @Override
      public void run2() throws CacheException {
        HARegionQueueDUnitTest test = new HARegionQueueDUnitTest();
        System.getProperties().put("QueueRemovalThreadWaitTime", "2000");
        cache = test.createCache();
        AttributesFactory factory = new AttributesFactory();
        factory.setScope(rscope);
        factory.setDataPolicy(DataPolicy.REPLICATE);
        HARegionQueueAttributes hrqa = new HARegionQueueAttributes();
        hrqa.setExpiryTime(5);
        try {
          if (createBlockingQueue) {
            hrq = HARegionQueue.getHARegionQueueInstance("testregion1", cache, hrqa,
                HARegionQueue.BLOCKING_HA_QUEUE, false, disabledClock());
          } else {
            hrq = HARegionQueue.getHARegionQueueInstance("testregion1", cache, hrqa,
                HARegionQueue.NON_BLOCKING_HA_QUEUE, false, disabledClock());
          }
        } catch (Exception e) {
          throw new AssertionError(e);
        }
        factory.addCacheListener(new CacheListenerAdapter() {
          @Override
          public void afterCreate(final EntryEvent event) {
            if (toCnt) {
              Conflatable conflatable = new ConflatableObject(event.getKey(), event.getNewValue(),
                  ((EntryEventImpl) event).getEventId(), false, event.getRegion().getFullPath());

              try {
                hrq.put(conflatable);
              } catch (Exception e) {
                fail("The put operation in queue did not succeed due to exception =", e);
              }
            }
          }

          @Override
          public void afterUpdate(final EntryEvent event) {
            if (toCnt) {
              Conflatable conflatable = new ConflatableObject(event.getKey(), event.getNewValue(),
                  ((EntryEventImpl) event).getEventId(), true, event.getRegion().getFullPath());

              try {
                hrq.put(conflatable);
              } catch (Exception e) {
                fail("The put operation in queue did not succeed due to exception =", e);
              }
            }
          }

        });

        cache.createRegion("test_region", factory.create());

      }
    };

    vm0.invoke(createRgnsAndQueues);
    vm1.invoke(createRgnsAndQueues);
    vm2.invoke(createRgnsAndQueues);
    vm3.invoke(createRgnsAndQueues);
    CacheSerializableRunnable spawnThreadsAndperformOps =
        new CacheSerializableRunnable("Spawn multiple threads which do various operations") {

          @Override
          public void run2() throws CacheException {
            opThreads = new RunOp[4 + 2 + 2 + 2];
            for (int i = 0; i < 4; ++i) {
              opThreads[i] = new RunOp(RunOp.PUT, i);
            }
            for (int i = 4; i < 6; ++i) {
              opThreads[i] = new RunOp(RunOp.PEEK, i);
            }

            for (int i = 6; i < 8; ++i) {
              opThreads[i] = new RunOp(RunOp.TAKE, i);
            }

            for (int i = 8; i < 10; ++i) {
              opThreads[i] = new RunOp(RunOp.TAKE, i);
            }

            for (final Thread opThread : opThreads) {
              opThread.start();
            }

          }
        };

    vm0.invokeAsync(spawnThreadsAndperformOps);
    vm1.invokeAsync(spawnThreadsAndperformOps);
    vm2.invokeAsync(spawnThreadsAndperformOps);
    vm3.invokeAsync(spawnThreadsAndperformOps);


    SerializableCallable guaranteeOperationsOccured =
        new SerializableCallable("Check Ops Occurred") {
          @Override
          public Object call() throws CacheException {
            if (opThreads == null) {
              return false;
            }
            for (final Thread opThread : opThreads) {
              if (((RunOp) opThread).getNumOpsPerformed() == 0) {
                return false;
              }
            }
            return true;
          }
        };

    await()
        .untilAsserted(() -> assertTrue((Boolean) vm0.invoke(guaranteeOperationsOccured)));
    await()
        .untilAsserted(() -> assertTrue((Boolean) vm1.invoke(guaranteeOperationsOccured)));
    await()
        .untilAsserted(() -> assertTrue((Boolean) vm2.invoke(guaranteeOperationsOccured)));
    await()
        .untilAsserted(() -> assertTrue((Boolean) vm3.invoke(guaranteeOperationsOccured)));

    // In case of blocking HARegionQueue do some extra puts so that the
    // blocking threads
    // are exited
    CacheSerializableRunnable toggleFlag =
        new CacheSerializableRunnable("Toggle the flag to signal end of threads") {
          @Override
          public void run2() throws CacheException {
            toCnt = false;
            if (createBlockingQueue) {
              try {
                for (int i = 0; i < 100; ++i) {
                  hrq.put(new ConflatableObject("1", "1", new EventID(new byte[] {1}, 100, i),
                      false, SEPARATOR + "x"));
                }
              } catch (Exception e) {
                throw new AssertionError(e);
              }
            }

          }
        };

    vm0.invokeAsync(toggleFlag);
    vm1.invokeAsync(toggleFlag);
    vm2.invokeAsync(toggleFlag);
    vm3.invokeAsync(toggleFlag);

    CacheSerializableRunnable joinWithThreads =
        new CacheSerializableRunnable("Join with the threads") {
          @Override
          public void run2() throws CacheException {
            for (final Thread opThread : opThreads) {

              if (opThread.isInterrupted()) {
                fail("Test failed because  thread encountered exception");
              }
              ThreadUtils.join(opThread, 30 * 1000);
            }
          }
        };

    vm0.invoke(joinWithThreads);
    vm1.invoke(joinWithThreads);
    vm2.invoke(joinWithThreads);
    vm3.invoke(joinWithThreads);
    System.getProperties().remove("QueueRemovalThreadWaitTime");
  }
Tests the Blokcing HARegionQueue by doing concurrent put /remove / take / peek , batch peek
operations in multiple regions. The test will have take/remove occuring in all the VMs. This
test is targetted to test for hang or exceptions in blocking queue.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/ha/HARegionQueueDUnitTest.java