GEODE-7465: Set eventProcessor to null in serial AEQ when it is stopped

geode-core/src/distributedTest/java/org/apache/geode/internal/cache/wan/asyncqueue/AsyncEventListenerDistributedTest.java

@@ -160,6 +160,31 @@ public void testSerialAsyncEventQueueSize() {
     assertThat(vm2.invoke(() -> getAsyncEventQueue().size())).isEqualTo(1000);
   }
 
+  @Test // serial, ReplicateRegion
+  public void testSerialAsyncEventQueueStopStart() {
+    vm0.invoke(this::createCache);
+    vm1.invoke(this::createCache);
+
+    vm0.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
+        100, dispatcherThreadCount, 100));
+    vm1.invoke(() -> createAsyncEventQueue(asyncEventQueueId, new SpyAsyncEventListener(), false,
+        100, dispatcherThreadCount, 100));
+
+    vm0.invoke(() -> createReplicateRegion(replicateRegionName, asyncEventQueueId));
+    vm1.invoke(() -> createReplicateRegion(replicateRegionName, asyncEventQueueId));
+
+    vm0.invoke(() -> getInternalGatewaySender().stop());
+    vm1.invoke(() -> getInternalGatewaySender().stop());
+
+    vm0.invoke(() -> doPuts(replicateRegionName, 10));
+
+    vm0.invoke(() -> getInternalGatewaySender().start());
+    vm1.invoke(() -> getInternalGatewaySender().start());
+
+    assertThat(vm0.invoke(() -> getAsyncEventQueue().size())).isEqualTo(0);
+    assertThat(vm1.invoke(() -> getAsyncEventQueue().size())).isEqualTo(0);
+  }
+
   @Test // serial, ReplicateRegion
   public void testReplicatedSerialAsyncEventQueue() {
     vm0.invoke(this::createCache);

geode-core/src/distributedTest/java/org/apache/geode/internal/cache/wan/asyncqueue/ParallelSerialAsyncEventListenerStopStartDistributedTest.java

@@ -0,0 +1,307 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.geode.internal.cache.wan.asyncqueue;
+
+import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT;
+import static org.apache.geode.cache.RegionShortcut.REPLICATE;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.VM.getCurrentVMNum;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.DiskStoreFactory;
+import org.apache.geode.cache.PartitionAttributesFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionFactory;
+import org.apache.geode.cache.asyncqueue.AsyncEvent;
+import org.apache.geode.cache.asyncqueue.AsyncEventListener;
+import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
+import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
+import org.apache.geode.cache.asyncqueue.internal.InternalAsyncEventQueue;
+import org.apache.geode.cache.wan.GatewaySender;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.test.dunit.AsyncInvocation;
+import org.apache.geode.test.dunit.ThreadUtils;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
+import org.apache.geode.test.dunit.rules.DistributedRule;
+import org.apache.geode.test.junit.categories.AEQTest;
+import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;
+
+/**
+ * Extracted from {@link AsyncEventListenerDistributedTest}.
+ */
+@Category(AEQTest.class)
+@SuppressWarnings("serial")
+public class ParallelSerialAsyncEventListenerStopStartDistributedTest implements Serializable {
+
+  @Rule
+  public DistributedRule distributedRule = new DistributedRule();
+
+  @Rule
+  public CacheRule cacheRule = new CacheRule();
+
+  @Rule
+  public DistributedRestoreSystemProperties restoreSystemProperties =
+      new DistributedRestoreSystemProperties();
+
+  @Rule
+  public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();
+
+  private String basePartitionedRegionName;
+  private String baseReplicateRegionName;
+  private String baseSerialAsyncEventQueueId;
+  private String baseParallelAsyncEventQueueId;
+
+  private VM vm0;
+  private VM vm1;
+  private VM vm2;
+  private VM vm3;
+
+  @Before
+  public void setUp() throws Exception {
+    vm0 = getVM(0);
+    vm1 = getVM(1);
+    vm2 = getVM(2);
+    vm3 = getVM(3);
+
+    String className = getClass().getSimpleName();
+    basePartitionedRegionName = className + "_PR_";
+    baseReplicateRegionName = className + "_RR_";
+    baseSerialAsyncEventQueueId = className + "_serialAEQ_";
+    baseParallelAsyncEventQueueId = className + "_parallelAEQ_";
+  }
+
+  /**
+   * Override as needed to add to the configuration, such as off-heap-memory-size.
+   */
+  protected Properties getDistributedSystemProperties() {
+    return new Properties();
+  }
+
+  /**
+   * Override as needed to add to the configuration, such as regionFactory.setOffHeap(boolean).
+   */
+  protected RegionFactory<?, ?> configureRegion(RegionFactory<?, ?> regionFactory) {
+    return regionFactory;
+  }
+
+  @Test
+  public void testStopStartPersistentParallelAndSerialAsyncEventQueues() {
+    // Create cache
+    vm0.invoke(() -> createCache());
+    vm1.invoke(() -> createCache());
+    vm2.invoke(() -> createCache());
+    vm3.invoke(() -> createCache());
+
+    // Create several serial and parallel AEQs
+    int numAEQs = 3;
+    vm0.invoke(() -> createAsyncEventQueues(numAEQs));
+    vm1.invoke(() -> createAsyncEventQueues(numAEQs));
+    vm2.invoke(() -> createAsyncEventQueues(numAEQs));
+    vm3.invoke(() -> createAsyncEventQueues(numAEQs));
+
+    // Create replicated and partitioned regions attached to those AEQs
+    int numRegions = 3;
+    vm0.invoke(() -> createRegions(numRegions));
+    vm1.invoke(() -> createRegions(numRegions));
+    vm2.invoke(() -> createRegions(numRegions));
+    vm3.invoke(() -> createRegions(numRegions));
+
+    // Do puts into all replicated and partitioned regions
+    vm0.invoke(() -> doPuts(numRegions, 1000));
+
+    // Wait for all AEQs to be empty
+    vm0.invoke(() -> waitForAsyncQueuesToEmpty());
+    vm1.invoke(() -> waitForAsyncQueuesToEmpty());
+    vm2.invoke(() -> waitForAsyncQueuesToEmpty());
+    vm3.invoke(() -> waitForAsyncQueuesToEmpty());
+
+    // Stop all AEQs
+    vm0.invoke(() -> stopAsyncQueues());
+    vm1.invoke(() -> stopAsyncQueues());
+    vm2.invoke(() -> stopAsyncQueues());
+    vm3.invoke(() -> stopAsyncQueues());
+
+    // Start all AEQs
+    AsyncInvocation startAeqsVm0 = vm0.invokeAsync(() -> startAsyncQueues());
+    AsyncInvocation startAeqsVm1 = vm1.invokeAsync(() -> startAsyncQueues());
+    AsyncInvocation startAeqsVm2 = vm2.invokeAsync(() -> startAsyncQueues());
+    AsyncInvocation startAeqsVm3 = vm3.invokeAsync(() -> startAsyncQueues());
+
+    // Wait for async tasks to complete
+    ThreadUtils.join(startAeqsVm0, 120 * 1000);
+    ThreadUtils.join(startAeqsVm1, 120 * 1000);
+    ThreadUtils.join(startAeqsVm2, 120 * 1000);
+    ThreadUtils.join(startAeqsVm3, 120 * 1000);
+  }
+
+  private void createAsyncEventQueues(int numAEQs) {
+    for (int i = 0; i < numAEQs; i++) {
+      createPersistentAsyncEventQueue(baseSerialAsyncEventQueueId + i,
+          createDiskStoreName(baseSerialAsyncEventQueueId + i), false);
+      createPersistentAsyncEventQueue(baseParallelAsyncEventQueueId + i,
+          createDiskStoreName(baseParallelAsyncEventQueueId + i), true);
+    }
+  }
+
+  private void createRegions(int numRegions) {
+    for (int i = 0; i < numRegions; i++) {
+      createReplicateRegion(baseReplicateRegionName + i, baseSerialAsyncEventQueueId + i);
+      createPartitionedRegion(basePartitionedRegionName + i, baseParallelAsyncEventQueueId + i);
+    }
+  }
+
+  private void waitForAsyncQueuesToEmpty() {
+    for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
+      await().until(() -> aeq.size() == 0);
+    }
+  }
+
+  private void stopAsyncQueues() {
+    for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
+      InternalAsyncEventQueue iaeq = (InternalAsyncEventQueue) aeq;
+      assertThat(iaeq.getSender().isRunning()).isTrue();
+      iaeq.stop();
+      assertThat(iaeq.getSender().isRunning()).isFalse();
+    }
+  }
+
+  private void startAsyncQueues() {
+    for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
+      GatewaySender sender = ((InternalAsyncEventQueue) aeq).getSender();
+      assertThat(sender.isRunning()).isFalse();
+      sender.start();
+      assertThat(sender.isRunning()).isTrue();
+    }
+  }
+
+  private Collection<AsyncEventQueue> getAsyncEventQueues() {
+    return getCache().getAsyncEventQueues().stream()
+        .sorted(Comparator.comparing(AsyncEventQueue::getId)).collect(Collectors.toList());
+  }
+
+  private InternalCache getCache() {
+    return cacheRule.getOrCreateCache(getDistributedSystemProperties());
+  }
+
+  private void createCache() {
+    cacheRule.createCache(getDistributedSystemProperties());
+  }
+
+  private void createPartitionedRegion(String regionName, String asyncEventQueueId) {
+    assertThat(regionName).isNotEmpty();
+    assertThat(asyncEventQueueId).isNotEmpty();
+
+    PartitionAttributesFactory<?, ?> partitionAttributesFactory = new PartitionAttributesFactory();
+
+    RegionFactory<?, ?> regionFactory = getCache().createRegionFactory(PARTITION_REDUNDANT);
+    regionFactory.addAsyncEventQueueId(asyncEventQueueId);
+    regionFactory.setPartitionAttributes(partitionAttributesFactory.create());
+
+    configureRegion(regionFactory).create(regionName);
+  }
+
+  private void createReplicateRegion(String regionName, String asyncEventQueueId) {
+    assertThat(regionName).isNotEmpty();
+    assertThat(asyncEventQueueId).isNotEmpty();
+
+    RegionFactory<?, ?> regionFactory = getCache().createRegionFactory(REPLICATE);
+    regionFactory.addAsyncEventQueueId(asyncEventQueueId);
+
+    configureRegion(regionFactory).create(regionName);
+  }
+
+  private void createDiskStore(String diskStoreName, String asyncEventQueueId) {
+    assertThat(diskStoreName).isNotEmpty();
+    assertThat(asyncEventQueueId).isNotEmpty();
+
+    File directory = createDirectory(createDiskStoreName(asyncEventQueueId));
+
+    DiskStoreFactory diskStoreFactory = getCache().createDiskStoreFactory();
+    diskStoreFactory.setDiskDirs(new File[] {directory});
+
+    diskStoreFactory.create(diskStoreName);
+  }
+
+  private File createDirectory(String name) {
+    assertThat(name).isNotEmpty();
+
+    File directory = new File(temporaryFolder.getRoot(), name);
+    if (!directory.exists()) {
+      try {
+        return temporaryFolder.newFolder(name);
+      } catch (IOException e) {
+        throw new UncheckedIOException(e);
+      }
+    }
+    return directory;
+  }
+
+  private String createDiskStoreName(String asyncEventQueueId) {
+    assertThat(asyncEventQueueId).isNotEmpty();
+
+    return asyncEventQueueId + "_disk_" + getCurrentVMNum();
+  }
+
+  private void createPersistentAsyncEventQueue(String asyncEventQueueId, String diskStoreName,
+      boolean isParallel) {
+    createDiskStore(diskStoreName, asyncEventQueueId);
+
+    AsyncEventQueueFactory asyncEventQueueFactory = getCache().createAsyncEventQueueFactory();
+    asyncEventQueueFactory.setDiskStoreName(diskStoreName);
+    asyncEventQueueFactory.setParallel(isParallel);
+    asyncEventQueueFactory.setPersistent(true);
+
+    asyncEventQueueFactory.create(asyncEventQueueId, new TestAsyncEventListener());
+  }
+
+  private void doPuts(int numRegions, int numPuts) {
+    for (int i = 0; i < numRegions; i++) {
+      Region<Integer, Integer> rr = getCache().getRegion(baseReplicateRegionName + i);
+      Region<Integer, Integer> pr = getCache().getRegion(basePartitionedRegionName + i);
+      for (int j = 0; j < numPuts; j++) {
+        rr.put(j, j);
+        pr.put(j, j);
+      }
+    }
+  }
+
+  private static class TestAsyncEventListener implements AsyncEventListener {
+
+    @Override
+    public synchronized boolean processEvents(List<AsyncEvent> events) {
+      return true;
+    }
+  }
+}

geode-core/src/integrationTest/java/org/apache/geode/cache/asyncqueue/internal/SerialAsyncEventQueueImplJUnitTest.java

@@ -68,4 +68,31 @@ public void testStopClearsStats() {
     assertEquals(0, queue.getStatistics().getEventsProcessedByPQRM());
   }
 
+  @Test
+  public void testStopStart() {
+    GatewaySenderAttributes attrs = new GatewaySenderAttributes();
+    attrs.id = AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX + "id";
+    SerialAsyncEventQueueImpl queue = new SerialAsyncEventQueueImpl(cache,
+        cache.getInternalDistributedSystem().getStatisticsManager(), cache.getStatisticsClock(),
+        attrs);
+    queue.getStatistics().incQueueSize(5);
+    queue.getStatistics().incSecondaryQueueSize(6);
+    queue.getStatistics().incTempQueueSize(10);
+    queue.getStatistics().incEventsProcessedByPQRM(3);
+
+    assertEquals(5, queue.getStatistics().getEventQueueSize());
+    assertEquals(6, queue.getStatistics().getSecondaryEventQueueSize());
+    assertEquals(10, queue.getStatistics().getTempEventQueueSize());
+    assertEquals(3, queue.getStatistics().getEventsProcessedByPQRM());
+
+    queue.start();
+    queue.stop();
+    queue.start();
+
+    assertEquals(0, queue.getStatistics().getEventQueueSize());
+    assertEquals(0, queue.getStatistics().getSecondaryEventQueueSize());
+    assertEquals(0, queue.getStatistics().getTempEventQueueSize());
+    assertEquals(0, queue.getStatistics().getEventsProcessedByPQRM());
+  }
+
 }