AbortingIndexerLock implements AsyncIndexerLock<SimpleToken> { @Override public SimpleToken lock(String asyncIndexerLane) { IndexStatsMBean mbean = getIndexStatsMBean(asyncIndexerLane); if (IndexStatsMBean.STATUS_RUNNING.equals(mbean.getStatus())){ log.info("Aborting current indexing run of async indexer for lane [{}]", asyncIndexerLane); } mbean.abortAndPause(); retry(mbean, TIMEOUT_SECONDS, 1000); log.info("Aborted and paused async indexer for lane [{}]", asyncIndexerLane); return new SimpleToken(asyncIndexerLane); } AbortingIndexerLock(AsyncIndexInfoService infoService);  AbortingIndexerLock(AsyncIndexInfoService infoService, Clock clock); @Override SimpleToken lock(String asyncIndexerLane); @Override void unlock(SimpleToken token); static final int TIMEOUT_SECONDS; }
@Test public void lockWithRetry() throws Exception{ AbortingIndexerLock lock = new AbortingIndexerLock(infoService); when(statsMBean.getStatus()) .thenReturn(IndexStatsMBean.STATUS_RUNNING) .thenReturn(IndexStatsMBean.STATUS_RUNNING) .thenReturn(IndexStatsMBean.STATUS_DONE); SimpleToken lockToken = lock.lock("async"); assertNotNull(lockToken); verify(statsMBean).abortAndPause(); verify(statsMBean, times(3)).getStatus(); }
@Test public void lockTimedout() throws Exception{ Clock.Virtual clock = new Clock.Virtual(); AbortingIndexerLock lock = new AbortingIndexerLock(infoService, clock); when(statsMBean.getStatus()) .thenReturn(IndexStatsMBean.STATUS_RUNNING) .thenReturn(IndexStatsMBean.STATUS_RUNNING) .then(invocation -> { clock.waitUntil(AbortingIndexerLock.TIMEOUT_SECONDS * 1000 + 1); return IndexStatsMBean.STATUS_RUNNING; }); try { lock.lock("async"); fail(); } catch (IllegalStateException ignore) { } verify(statsMBean).abortAndPause(); verify(statsMBean, times(3)).getStatus(); }