@Test
  public void testParallelGatewaySenderConcurrentPutClearNoOffheapOrphans() {
    MemberVM locator = clusterStartupRule.startLocatorVM(1, new Properties());
    Properties properties = new Properties();
    properties.put(OFF_HEAP_MEMORY_SIZE_NAME, "100");
    MemberVM server = clusterStartupRule.startServerVM(2, properties, locator.getPort());
    final String regionName = "portfolios";
    final String gatewaySenderId = "ln";

    server.invoke(() -> {
      addIgnoredException("could not get remote locator");
      InternalCache cache = ClusterStartupRule.getCache();
      GatewaySender sender =
          cache.createGatewaySenderFactory().setParallel(true).create(gatewaySenderId, 1);
      Region<Object, Object> userRegion =
          cache.createRegionFactory(RegionShortcut.PARTITION).setOffHeap(true)
              .addGatewaySenderId("ln").create(regionName);
      PartitionedRegion shadowRegion = (PartitionedRegion) ((AbstractGatewaySender) sender)
          .getEventProcessor().getQueue().getRegion();
      @SuppressWarnings("unchecked")
      CacheWriter<Object, Object> mockCacheWriter = mock(CacheWriter.class);
      CountDownLatch cacheWriterLatch = new CountDownLatch(1);
      CountDownLatch shadowRegionClearLatch = new CountDownLatch(1);

      doAnswer(invocation -> {
        // The cache writer is invoked between when the region entry is created with value of
        // Token.REMOVED_PHASE_1 and when it is replaced with the actual GatewaySenderEvent.
        // We use this hook to trigger the clear logic when the token is still in the
        // region entry.
        cacheWriterLatch.countDown();
        // Wait until the clear is complete before putting the actual GatewaySenderEvent in the
        // region entry.
        shadowRegionClearLatch.await();
        return null;
      }).when(mockCacheWriter).beforeCreate(any());

      shadowRegion.setCacheWriter(mockCacheWriter);

      ExecutorService service = Executors.newFixedThreadPool(2);

      List<Callable<Object>> callables = new ArrayList<>();

      // In one thread, we are going to put some test data in the user region,
      // which will eventually put the GatewaySenderEvent into the shadow region
      callables.add(Executors.callable(() -> {
        try {
          userRegion.put("testKey", "testValue");
        } catch (Exception ex) {
          throw new RuntimeException(ex);
        }
      }));

      // In another thread, we are clear the shadow region's buckets. If the region entry is a
      // Token.REMOVED_PHASE_1 at this time, a leak is possible. We can guarantee that the
      // Token.REMOVED_PHASE_1 is present by using the mocked cache writer defined
      // above.
      callables.add(Executors.callable(() -> {
        try {
          OffHeapClearRequired.doWithOffHeapClear(() -> {
            // Wait for the cache writer to be invoked to release this countdown latch.
            // This guarantees that the region entry will contain a Token.REMOVED_PHASE_1.
            try {
              cacheWriterLatch.await();
            } catch (InterruptedException e) {
              throw new TestException(
                  "Thread was interrupted while waiting for mocked cache writer to be invoked");
            }

            clearShadowBucketRegions(shadowRegion);

            // Signal to the cache writer that the clear is complete and the put can continue.
            shadowRegionClearLatch.countDown();
          });
        } catch (Exception ex) {
          throw new RuntimeException(ex);
        }
      }));

      List<Future<Object>> futures = service.invokeAll(callables, 10, TimeUnit.SECONDS);

      for (Future<Object> future : futures) {
        try {
          future.get();
        } catch (Exception ex) {
          throw new TestException(
              "Exception thrown while executing put and clear concurrently",
              ex);
        }
      }

      userRegion.close();

      await("Waiting for off-heap to be freed").until(
          () -> 0 == ((MemoryAllocatorImpl) cache.getOffHeapStore()).getOrphans(cache).size());
    });
  }
Since the sender is attached to a region and in use, it can not be destroyed. Hence, exception
is thrown by the sender API.
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDistributedTest.java