@Test public void testFindMissingKeys() throws Exception { setupCloudStore(false, true, defaultCacheLimit, true); int count = 10; List<StoreKey> keys = new ArrayList<>(); Map<String, CloudBlobMetadata> metadataMap = new HashMap<>(); for (int j = 0; j < count; j++) { BlobId existentBlobId = getUniqueId(refAccountId, refContainerId, false, partitionId); keys.add(existentBlobId); metadataMap.put(existentBlobId.getID(), new CloudBlobMetadata(existentBlobId, operationTime, Utils.Infinite_Time, 1024, CloudBlobMetadata.EncryptionOrigin.ROUTER)); BlobId nonexistentBlobId = getUniqueId(refAccountId, refContainerId, false, partitionId); keys.add(nonexistentBlobId); } when(dest.getBlobMetadata(anyList())).thenReturn(metadataMap); Set<StoreKey> missingKeys = store.findMissingKeys(keys); verify(dest).getBlobMetadata(anyList()); int expectedLookups = keys.size(); int expectedHits = 0; verifyCacheHits(expectedLookups, expectedHits); assertEquals("Wrong number of missing keys", count, missingKeys.size()); if (isVcr) { for (StoreKey storeKey : keys) { store.addToCache(storeKey.getID(), (short) 0, CloudBlobStore.BlobState.CREATED); } missingKeys = store.findMissingKeys(keys); assertTrue("Expected no missing keys", missingKeys.isEmpty()); expectedLookups += keys.size(); expectedHits += keys.size(); verifyCacheHits(expectedLookups, expectedHits); verify(dest).getBlobMetadata(anyList()); } }
@Override public Set<StoreKey> findMissingKeys(List<StoreKey> keys) throws StoreException { checkStarted(); List<BlobId> blobIdQueryList = keys.stream() .filter(key -> !checkCacheState(key.getID())) .map(key -> (BlobId) key) .collect(Collectors.toList()); if (blobIdQueryList.isEmpty()) { return Collections.emptySet(); } try { Set<String> foundSet = requestAgent.doWithRetries(() -> cloudDestination.getBlobMetadata(blobIdQueryList), "FindMissingKeys", partitionId.toPathString()).keySet(); return keys.stream() .filter(key -> !foundSet.contains(key.getID())) .filter(key -> !recentBlobCache.containsKey(key.getID())) .collect(Collectors.toSet()); } catch (CloudStorageException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } }
CloudBlobStore implements Store { @Override public Set<StoreKey> findMissingKeys(List<StoreKey> keys) throws StoreException { checkStarted(); List<BlobId> blobIdQueryList = keys.stream() .filter(key -> !checkCacheState(key.getID())) .map(key -> (BlobId) key) .collect(Collectors.toList()); if (blobIdQueryList.isEmpty()) { return Collections.emptySet(); } try { Set<String> foundSet = requestAgent.doWithRetries(() -> cloudDestination.getBlobMetadata(blobIdQueryList), "FindMissingKeys", partitionId.toPathString()).keySet(); return keys.stream() .filter(key -> !foundSet.contains(key.getID())) .filter(key -> !recentBlobCache.containsKey(key.getID())) .collect(Collectors.toSet()); } catch (CloudStorageException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } }
CloudBlobStore implements Store { @Override public Set<StoreKey> findMissingKeys(List<StoreKey> keys) throws StoreException { checkStarted(); List<BlobId> blobIdQueryList = keys.stream() .filter(key -> !checkCacheState(key.getID())) .map(key -> (BlobId) key) .collect(Collectors.toList()); if (blobIdQueryList.isEmpty()) { return Collections.emptySet(); } try { Set<String> foundSet = requestAgent.doWithRetries(() -> cloudDestination.getBlobMetadata(blobIdQueryList), "FindMissingKeys", partitionId.toPathString()).keySet(); return keys.stream() .filter(key -> !foundSet.contains(key.getID())) .filter(key -> !recentBlobCache.containsKey(key.getID())) .collect(Collectors.toSet()); } catch (CloudStorageException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } CloudBlobStore(VerifiableProperties properties, PartitionId partitionId, CloudDestination cloudDestination,
      ClusterMap clusterMap, VcrMetrics vcrMetrics); }
CloudBlobStore implements Store { @Override public Set<StoreKey> findMissingKeys(List<StoreKey> keys) throws StoreException { checkStarted(); List<BlobId> blobIdQueryList = keys.stream() .filter(key -> !checkCacheState(key.getID())) .map(key -> (BlobId) key) .collect(Collectors.toList()); if (blobIdQueryList.isEmpty()) { return Collections.emptySet(); } try { Set<String> foundSet = requestAgent.doWithRetries(() -> cloudDestination.getBlobMetadata(blobIdQueryList), "FindMissingKeys", partitionId.toPathString()).keySet(); return keys.stream() .filter(key -> !foundSet.contains(key.getID())) .filter(key -> !recentBlobCache.containsKey(key.getID())) .collect(Collectors.toSet()); } catch (CloudStorageException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } CloudBlobStore(VerifiableProperties properties, PartitionId partitionId, CloudDestination cloudDestination,
      ClusterMap clusterMap, VcrMetrics vcrMetrics); @Override void start(); @Override StoreInfo get(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions); @Override void put(MessageWriteSet messageSetToWrite); @Override void delete(List<MessageInfo> infos); @Override short undelete(MessageInfo info); @Override void updateTtl(List<MessageInfo> infos); @Override FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries); @Override Set<StoreKey> findMissingKeys(List<StoreKey> keys); @Override MessageInfo findKey(StoreKey key); @Override StoreStats getStoreStats(); @Override boolean isKeyDeleted(StoreKey key); @Override long getSizeInBytes(); @Override boolean isEmpty(); @Override boolean isBootstrapInProgress(); @Override boolean isDecommissionInProgress(); @Override void completeBootstrap(); @Override void setCurrentState(ReplicaState state); @Override ReplicaState getCurrentState(); @Override long getEndPositionOfLastPut(); @Override boolean recoverFromDecommission(); @Override boolean disabledOnError(); @Override void shutdown(); @Override boolean isStarted(); @Override String toString(); }
CloudBlobStore implements Store { @Override public Set<StoreKey> findMissingKeys(List<StoreKey> keys) throws StoreException { checkStarted(); List<BlobId> blobIdQueryList = keys.stream() .filter(key -> !checkCacheState(key.getID())) .map(key -> (BlobId) key) .collect(Collectors.toList()); if (blobIdQueryList.isEmpty()) { return Collections.emptySet(); } try { Set<String> foundSet = requestAgent.doWithRetries(() -> cloudDestination.getBlobMetadata(blobIdQueryList), "FindMissingKeys", partitionId.toPathString()).keySet(); return keys.stream() .filter(key -> !foundSet.contains(key.getID())) .filter(key -> !recentBlobCache.containsKey(key.getID())) .collect(Collectors.toSet()); } catch (CloudStorageException ex) { throw new StoreException(ex, StoreErrorCodes.IOError); } } CloudBlobStore(VerifiableProperties properties, PartitionId partitionId, CloudDestination cloudDestination,
      ClusterMap clusterMap, VcrMetrics vcrMetrics); @Override void start(); @Override StoreInfo get(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions); @Override void put(MessageWriteSet messageSetToWrite); @Override void delete(List<MessageInfo> infos); @Override short undelete(MessageInfo info); @Override void updateTtl(List<MessageInfo> infos); @Override FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries); @Override Set<StoreKey> findMissingKeys(List<StoreKey> keys); @Override MessageInfo findKey(StoreKey key); @Override StoreStats getStoreStats(); @Override boolean isKeyDeleted(StoreKey key); @Override long getSizeInBytes(); @Override boolean isEmpty(); @Override boolean isBootstrapInProgress(); @Override boolean isDecommissionInProgress(); @Override void completeBootstrap(); @Override void setCurrentState(ReplicaState state); @Override ReplicaState getCurrentState(); @Override long getEndPositionOfLastPut(); @Override boolean recoverFromDecommission(); @Override boolean disabledOnError(); @Override void shutdown(); @Override boolean isStarted(); @Override String toString();  }