SerialAsyncEventQueueImpl extends AbstractGatewaySender { @Override public void stop() { if (logger.isDebugEnabled()) { logger.debug("Stopping Gateway Sender : {}", this); } this.getLifeCycleLock().writeLock().lock(); try { stopProcessing(); stompProxyDead(); for (AsyncEventListener listener : this.listeners) { listener.close(); } logger.info("Stopped {}", this); clearTempEventsAfterSenderStopped(); } finally { this.getLifeCycleLock().writeLock().unlock(); } if (this.isPrimary()) { try { DistributedLockService.destroy(getSenderAdvisor().getDLockServiceName()); } catch (IllegalArgumentException e) { } } if (getQueues() != null && !getQueues().isEmpty()) { for (RegionQueue q : getQueues()) { ((SerialGatewaySenderQueue) q).cleanUp(); } } this.setIsPrimary(false); try { new UpdateAttributesProcessor(this).distribute(false); } catch (CancelException e) { } Thread lockObtainingThread = getSenderAdvisor().getLockObtainingThread(); if (lockObtainingThread != null && lockObtainingThread.isAlive()) { try { lockObtainingThread.join(3000); } catch (InterruptedException ex) { Thread.currentThread().interrupt(); } if (lockObtainingThread.isAlive()) { logger.info("Could not stop lock obtaining thread during gateway sender stop"); } } InternalDistributedSystem system = (InternalDistributedSystem) this.cache.getDistributedSystem(); system.handleResourceEvent(ResourceEvent.GATEWAYSENDER_STOP, this); } SerialAsyncEventQueueImpl(InternalCache cache, StatisticsFactory statisticsFactory,
      StatisticsClock statisticsClock, GatewaySenderAttributes attrs); @Override void start(); @Override void startWithCleanQueue(); @Override void stop(); @Override String toString(); @Override void fillInProfile(Profile profile); @Override void setModifiedEventId(EntryEventImpl clonedEvent);  }
@Test public void whenStoppedShouldResetTheEventProcessor() { serialAsyncEventQueue = createSerialAsyncEventQueueImplSpy(); serialAsyncEventQueue.stop(); assertThat(serialAsyncEventQueue.getEventProcessor()).isNull(); }