CompletableFuture<Map<UUID, Long>> getBucketOffsets(DirectSegmentAccess segment, Collection<UUID> hashes, TimeoutTimer timer) {
        Exceptions.checkNotClosed(this.closed.get(), this);

        if (hashes.isEmpty()) {
            // Nothing to search.
            return CompletableFuture.completedFuture(Collections.emptyMap());
        }

        // Find those keys which already exist in the cache. The same hash may occur multiple times, but this process
        // helps dedupe it.
        val result = new HashMap<UUID, Long>();
        val toLookup = new ArrayList<UUID>();
        for (UUID hash : hashes) {
            if (result.containsKey(hash)) {
                // This KeyHash has already been processed.
                continue;
            }

            val existingValue = this.cache.get(segment.getSegmentId(), hash);
            if (existingValue == null) {
                // Key Hash does not exist in the cache (it may or may not exist at all). Add a placeholder and keep
                // track of it so we can look it up.
                result.put(hash, TableKey.NOT_EXISTS);
                toLookup.add(hash);
            } else if (!existingValue.isRemoval()) {
                // Key Hash exists.
                result.put(hash, existingValue.getSegmentOffset());
            } else {
                long backpointerOffset = this.cache.getBackpointer(segment.getSegmentId(), existingValue.getSegmentOffset());
                if (backpointerOffset < 0) {
                    // Key Hash does not exist in the cache. Queue it up for lookup.
                    result.put(hash, TableKey.NOT_EXISTS);
                    toLookup.add(hash);
                } else {
                    // Key Hash (Table Bucket) has been created/updated recently, however it also had a removal, as such
                    // we are pointing to the last update, but there are other entries for this Bucket that may be of interest
                    // to the caller.
                    result.put(hash, existingValue.getSegmentOffset());
                }
            }
        }

        if (toLookup.isEmpty()) {
            // Full cache hit.
            return CompletableFuture.completedFuture(result);
        } else {
            // Fetch information for missing hashes.
            return this.recoveryTracker.waitIfNeeded(segment, () -> getBucketOffsetFromSegment(segment, result, toLookup, timer));
        }
    }
CompletableFuture < Map < UUID , Long > > getBucketOffsets ( DirectSegmentAccess segment , Collection < UUID > hashes , TimeoutTimer timer ) { Exceptions . checkNotClosed ( this . closed . get ( ) , this ) ; if ( hashes . isEmpty ( ) ) { return CompletableFuture . completedFuture ( Collections . emptyMap ( ) ) ; } val result = new HashMap < UUID , Long > ( ) ; val toLookup = new ArrayList < UUID > ( ) ; for ( UUID hash : hashes ) { if ( result . containsKey ( hash ) ) { continue ; } val existingValue = this . cache . get ( segment . getSegmentId ( ) , hash ) ; if ( existingValue == null ) { result . put ( hash , TableKey . NOT_EXISTS ) ; toLookup . add ( hash ) ; } else if ( ! existingValue . isRemoval ( ) ) { result . put ( hash , existingValue . getSegmentOffset ( ) ) ; } else { long backpointerOffset = this . cache . getBackpointer ( segment . getSegmentId ( ) , existingValue . getSegmentOffset ( ) ) ; if ( backpointerOffset < 0 ) { result . put ( hash , TableKey . NOT_EXISTS ) ; toLookup . add ( hash ) ; } else { result . put ( hash , existingValue . getSegmentOffset ( ) ) ; } } } if ( toLookup . isEmpty ( ) ) { return CompletableFuture . completedFuture ( result ) ; } else { return this . recoveryTracker . waitIfNeeded ( segment , ( ) -> getBucketOffsetFromSegment ( segment , result , toLookup , timer ) ) ; } }
Finds the Last Bucket Offsets for the given KeyHashes.

@param segment Segment to look up Bucket Offsets for.
@param hashes  A Collection of Key Hashes to identify the Buckets.
@param timer   Timer for the operation.
@return A CompletableFuture that, when completed, will contain the sought Offsets indexed by their corresponding
Key Hash. If a particular bucket does not exist, its corresponding Key Hash will have a {@link TableKey#NOT_EXISTS}
associated with it.