OwnershipCache { public Optional<String> resolveExecutor(final int blockId) { final String ownerId = blockOwnerArray.get(blockId); if (ownerId.equals(localExecutorId)) { return Optional.empty(); } else { return Optional.of(ownerId); } } @Inject private  OwnershipCache(@Parameter(NumTotalBlocks.class) final int numTotalBlocks,
                         @Parameter(TableIdentifier.class) final String tableId,
                         @Parameter(ExecutorIdentifier.class) final String executorId,
                         @Parameter(DriverIdentifier.class) final String driverId,
                         final InjectionFuture<MessageSender> msgSenderFuture); void init(final List<String> blockOwners); Optional<String> resolveExecutor(final int blockId); Pair<Optional<String>, Lock> resolveExecutorWithLock(final int blockId); List<Integer> getCurrentLocalBlockIds(); void update(final int blockId, final String oldOwnerId, final String newOwnerId); Lock holdWriteLock(final int blockId); void completeAllOngoingSync(); void syncUnassociation(final long opId, final String executorId); void allowAccessToBlock(final int blockId);  }
@Test public void testMultipleOwnershipCaches() throws InjectionException { final int numTotalBlocks = 1024; final int numExecutors = 4; final Map<String, OwnershipCache> ownershipCaches = newOwnershipCaches(numExecutors, numTotalBlocks); for (int blockId = 0; blockId < numTotalBlocks; blockId++) { String firstAnswer = null; boolean localStoreFound = false; for (final Map.Entry<String, OwnershipCache> entry : ownershipCaches.entrySet()) { final String localExecutorId = entry.getKey(); final OwnershipCache ownershipCache = entry.getValue(); final Optional<String> executorId = ownershipCache.resolveExecutor(blockId); final String targetExecutorId; if (!executorId.isPresent()) { assertFalse("Block should belong to only one store", localStoreFound); localStoreFound = true; targetExecutorId = localExecutorId; } else { targetExecutorId = executorId.get(); } if (firstAnswer == null) { firstAnswer = targetExecutorId; } else { assertEquals("Ownership caches should give the same memory store id for the same block", firstAnswer, targetExecutorId); } } } }