IndexRepositoryFactory { protected IndexRepository finishComputingRepository(Integer bucketId, LuceneSerializer serializer, PartitionedRegion userRegion, IndexRepository oldRepository, InternalLuceneIndex index) { LuceneIndexForPartitionedRegion indexForPR = (LuceneIndexForPartitionedRegion) index; final PartitionedRegion fileRegion = indexForPR.getFileAndChunkRegion(); BucketRegion fileAndChunkBucket = getMatchingBucket(fileRegion, bucketId); BucketRegion dataBucket = getMatchingBucket(userRegion, bucketId); boolean success = false; if (fileAndChunkBucket == null) { if (oldRepository != null) { oldRepository.cleanup(); } return null; } if (!fileAndChunkBucket.getBucketAdvisor().isPrimary()) { if (oldRepository != null) { oldRepository.cleanup(); } return null; } if (oldRepository != null && !oldRepository.isClosed()) { return oldRepository; } if (oldRepository != null) { oldRepository.cleanup(); } DistributedLockService lockService = getLockService(); String lockName = getLockName(fileAndChunkBucket); while (!lockService.lock(lockName, 100, -1)) { if (!fileAndChunkBucket.getBucketAdvisor().isPrimary()) { return null; } } final IndexRepository repo; InternalCache cache = (InternalCache) userRegion.getRegionService(); boolean initialPdxReadSerializedFlag = cache.getPdxReadSerializedOverride(); cache.setPdxReadSerializedOverride(true); try { IndexWriter writer = buildIndexWriter(bucketId, fileAndChunkBucket, indexForPR); repo = new IndexRepositoryImpl(fileAndChunkBucket, writer, serializer, indexForPR.getIndexStats(), dataBucket, lockService, lockName, indexForPR); success = false; if (null != fileRegion.get(APACHE_GEODE_INDEX_COMPLETE, bucketId)) { success = true; return repo; } else { success = reindexUserDataRegion(bucketId, userRegion, fileRegion, dataBucket, repo); } return repo; } catch (IOException e) { logger.warn("Exception thrown while constructing Lucene Index for bucket:" + bucketId + " for file region:" + fileAndChunkBucket.getFullPath(), e); return null; } catch (CacheClosedException e) { logger.info("CacheClosedException thrown while constructing Lucene Index for bucket:" + bucketId + " for file region:" + fileAndChunkBucket.getFullPath()); throw e; } finally { if (!success) { lockService.unlock(lockName); } cache.setPdxReadSerializedOverride(initialPdxReadSerializedFlag); } } IndexRepositoryFactory(); IndexRepository computeIndexRepository(final Integer bucketId, LuceneSerializer serializer,
      InternalLuceneIndex index, PartitionedRegion userRegion, final IndexRepository oldRepository,
      PartitionedRepositoryManager partitionedRepositoryManager); static final String FILE_REGION_LOCK_FOR_BUCKET_ID; static final String APACHE_GEODE_INDEX_COMPLETE; }
@Test public void finishComputingRepositoryShouldReturnNullAndCleanOldRepositoryWhenFileAndChunkBucketIsNotPrimary() { when(fileAndChunkBucketAdvisor.isPrimary()).thenReturn(false); IndexRepository indexRepository = indexRepositoryFactory.finishComputingRepository(0, serializer, userRegion, oldRepository, luceneIndex); assertThat(indexRepository).isNull(); verify(oldRepository).cleanup(); }
@Test public void finishComputingRepositoryShouldReturnOldRepositoryWhenNotNullAndNotClosed() { when(oldRepository.isClosed()).thenReturn(false); when(fileAndChunkBucketAdvisor.isPrimary()).thenReturn(true); IndexRepository indexRepository = indexRepositoryFactory.finishComputingRepository(0, serializer, userRegion, oldRepository, luceneIndex); assertThat(indexRepository).isNotNull(); assertThat(indexRepository).isSameAs(oldRepository); }
@Test public void finishComputingRepositoryShouldReturnNullWhenLockCanNotBeAcquiredAndFileAndChunkBucketIsNotPrimary() { when(oldRepository.isClosed()).thenReturn(true); when(fileAndChunkBucketAdvisor.isPrimary()).thenReturn(true).thenReturn(false); when(distributedLockService.lock(any(), anyLong(), anyLong())).thenReturn(false); IndexRepository indexRepository = indexRepositoryFactory.finishComputingRepository(0, serializer, userRegion, oldRepository, luceneIndex); assertThat(indexRepository).isNull(); }