class FetchingSnapshotShardSizeRunnable extends AbstractRunnable {

        private final SnapshotShard snapshotShard;
        private boolean removed;

        FetchingSnapshotShardSizeRunnable(SnapshotShard snapshotShard) {
            super();
            this.snapshotShard = snapshotShard;
            this.removed = false;
        }

        @Override
        protected void doRun() throws Exception {
            final RepositoriesService repositories = repositoriesService.get();
            assert repositories != null;
            final Repository repository = repositories.repository(snapshotShard.snapshot.getRepository());

            LOGGER.debug("fetching snapshot shard size for {}", snapshotShard);

            repository.getShardSnapshotStatus(
                snapshotShard.snapshot().getSnapshotId(),
                snapshotShard.index(),
                snapshotShard.shardId()).thenAccept(shardSnapshotStatus -> {
                    final long snapshotShardSize = shardSnapshotStatus == null ? 0
                            : shardSnapshotStatus.totalSize();
                    LOGGER.debug("snapshot shard size for {}: {} bytes", snapshotShard, snapshotShardSize);

                    boolean updated = false;
                    synchronized (mutex) {
                        removed = unknownSnapshotShards.remove(snapshotShard);
                        assert removed : "snapshot shard to remove does not exist " + snapshotShardSize;
                        if (isMaster) {
                            final ImmutableOpenMap.Builder<SnapshotShard, Long> newSnapshotShardSizes =
                                ImmutableOpenMap.builder(knownSnapshotShards);
                            updated = newSnapshotShardSizes.put(snapshotShard, snapshotShardSize) == null;
                            assert updated : "snapshot shard size already exists for " + snapshotShard;
                            knownSnapshotShards = newSnapshotShardSizes.build();
                        }
                        activeFetches -= 1;
                        assert invariant();
                    }
                    if (updated) {
                        rerouteService.get().reroute("snapshot shard size updated", Priority.HIGH, REROUTE_LISTENER);
                    }
                });
        }

        @Override
        public void onFailure(Exception e) {
            LOGGER.warn(() -> new ParameterizedMessage("failed to retrieve shard size for {}", snapshotShard), e);
            boolean failed = false;
            synchronized (mutex) {
                if (isMaster) {
                    failed = failedSnapshotShards.add(snapshotShard);
                    assert failed : "snapshot shard size already failed for " + snapshotShard;
                }
                if (removed == false) {
                    unknownSnapshotShards.remove(snapshotShard);
                }
                activeFetches -= 1;
                assert invariant();
            }
            if (failed) {
                rerouteService.get().reroute("snapshot shard size failed", Priority.HIGH, REROUTE_LISTENER);
            }
        }

        @Override
        public void onAfter() {
            fetchNextSnapshotShard();
        }
    }

    private void cleanUpSnapshotShardSizes(Set<SnapshotShard> requiredSnapshotShards) {
        assert Thread.holdsLock(mutex);
        ImmutableOpenMap.Builder<SnapshotShard, Long> newSnapshotShardSizes = null;
        for (ObjectCursor<SnapshotShard> shard : knownSnapshotShards.keys()) {
            if (requiredSnapshotShards.contains(shard.value) == false) {
                if (newSnapshotShardSizes == null) {
                    newSnapshotShardSizes = ImmutableOpenMap.builder(knownSnapshotShards);
                }
                newSnapshotShardSizes.remove(shard.value);
            }
        }
        if (newSnapshotShardSizes != null) {
            knownSnapshotShards = newSnapshotShardSizes.build();
        }
        failedSnapshotShards.retainAll(requiredSnapshotShards);
    }

    private boolean invariant() {<FILL_FUNCTION_BODY>

        assert Thread.holdsLock(mutex);
        assert activeFetches >= 0 : "active fetches should be greater than or equal to zero but got: " + activeFetches;
        assert activeFetches <= maxConcurrentFetches : activeFetches + " <= " + maxConcurrentFetches;
        for (ObjectCursor<SnapshotShard> cursor : knownSnapshotShards.keys()) {
            assert unknownSnapshotShards.contains(cursor.value) == false : "cannot be known and unknown at same time: " + cursor.value;
            assert failedSnapshotShards.contains(cursor.value) == false : "cannot be known and failed at same time: " + cursor.value;
        }
        for (SnapshotShard shard : unknownSnapshotShards) {
            assert knownSnapshotShards.keys().contains(shard) == false : "cannot be unknown and known at same time: " + shard;
            assert failedSnapshotShards.contains(shard) == false : "cannot be unknown and failed at same time: " + shard;
        }
        for (SnapshotShard shard : failedSnapshotShards) {
            assert knownSnapshotShards.keys().contains(shard) == false : "cannot be failed and known at same time: " + shard;
            assert unknownSnapshotShards.contains(shard) == false : "cannot be failed and unknown at same time: " + shard;
        }
        return true;
    
<no_super_class>