class BlobIndicesService implements IndexEventListener {

    private static final Logger LOGGER = LogManager.getLogger(BlobIndicesService.class);

    public static final Setting<Boolean> SETTING_INDEX_BLOBS_ENABLED = Setting.boolSetting(
        "index.blobs.enabled", false, Setting.Property.IndexScope);
    public static final Setting<String> SETTING_INDEX_BLOBS_PATH = Setting.simpleString(
        "index.blobs.path", Setting.Property.IndexScope);
    public static final Setting<String> SETTING_BLOBS_PATH = Setting.simpleString(
        "blobs.path", Setting.Property.NodeScope);

    private final ClusterService clusterService;

    final Map<String, BlobIndex> indices = new ConcurrentHashMap<>();

    @Nullable
    private final Path globalBlobPath;

    public BlobIndicesService(Settings settings, ClusterService clusterService) {
        this.clusterService = clusterService;
        globalBlobPath = getGlobalBlobPath(settings);
    }

    @Nullable
    public static Path getGlobalBlobPath(Settings settings) {
        String customGlobalBlobPathSetting = SETTING_BLOBS_PATH.get(settings);
        if (Strings.isNullOrEmpty(customGlobalBlobPathSetting)) {
            return null;
        }
        Path globalBlobPath = PathUtils.get(customGlobalBlobPathSetting);
        ensureExistsAndWritable(globalBlobPath);
        return globalBlobPath;
    }


    @Override
    public void afterIndexCreated(IndexService indexService) {
        String indexName = indexService.index().getName();
        if (isBlobIndex(indexName)) {
            BlobIndex oldBlobIndex = indices.put(indexName, new BlobIndex(LOGGER, globalBlobPath));
            assert oldBlobIndex == null : "There must not be an index present if a new index is created";
        }
    }

    @Override
    public void afterIndexRemoved(Index index,
                                  IndexSettings indexSettings,
                                  IndicesClusterStateService.AllocatedIndices.IndexRemovalReason reason) {
        String indexName = index.getName();
        if (isBlobIndex(indexName)) {
            BlobIndex blobIndex = indices.remove(indexName);
            assert blobIndex != null : "BlobIndex not found on afterIndexDeleted";
        }
    }

    @Override
    public void afterIndexShardCreated(IndexShard indexShard) {<FILL_FUNCTION_BODY>}

    @Override
    public void indexShardStateChanged(IndexShard indexShard,
                                       @Nullable IndexShardState previousState,
                                       IndexShardState currentState,
                                       @Nullable String reason) {
        if (currentState == IndexShardState.POST_RECOVERY) {
            String index = indexShard.shardId().getIndexName();
            if (isBlobIndex(index)) {
                BlobIndex blobIndex = indices.get(index);
                blobIndex.initializeShard(indexShard);
            }
        }
    }

    @Override
    public void afterIndexShardDeleted(ShardId shardId, Settings indexSettings) {
        String index = shardId.getIndexName();
        if (isBlobIndex(index)) {
            BlobIndex blobIndex = indices.get(index);
            if (blobIndex != null) {
                blobIndex.removeShard(shardId);
            }
        }
    }

    @Nullable
    public BlobShard blobShard(ShardId shardId) {
        BlobIndex blobIndex = indices.get(shardId.getIndexName());
        if (blobIndex == null) {
            return null;
        }
        return blobIndex.getShard(shardId.id());
    }

    public BlobShard blobShardSafe(ShardId shardId) {
        String index = shardId.getIndexName();
        if (isBlobIndex(index)) {
            BlobShard blobShard = blobShard(shardId);
            if (blobShard == null) {
                throw new ShardNotFoundException(shardId);
            }
            return blobShard;
        }
        throw new BlobsDisabledException(shardId.getIndex());
    }

    public BlobShard localBlobShard(String index, String digest) {
        return blobShardSafe(localShardId(index, digest));
    }

    private ShardId localShardId(String index, String digest) {
        ShardIterator si = clusterService.operationRouting().getShards(
            clusterService.state(), index, null, digest, "_only_local");
        return si.shardId();
    }

    static boolean ensureExistsAndWritable(Path blobsPath) {
        if (Files.exists(blobsPath)) {
            if (!Files.isDirectory(blobsPath)) {
                throw new SettingsException(
                    String.format(Locale.ENGLISH, "blobs path '%s' is a file, must be a directory", blobsPath));
            }
            if (!Files.isWritable(blobsPath)) {
                throw new SettingsException(
                    String.format(Locale.ENGLISH, "blobs path '%s' is not writable", blobsPath));
            }
        } else {
            try {
                Files.createDirectories(blobsPath);
            } catch (IOException e) {
                throw new SettingsException(
                    String.format(Locale.ENGLISH, "blobs path '%s' could not be created", blobsPath));
            }
        }
        return true;
    }
}

        String index = indexShard.shardId().getIndexName();
        if (isBlobIndex(index)) {
            BlobIndex blobIndex = indices.get(index);
            assert blobIndex != null : "blobIndex must exists if a shard is created in it";
            blobIndex.createShard(indexShard);
        }
    
<no_super_class>