@Override
    public <T> Mono<Void> ensureIndexes(Class<T> clazz, List<IndexModel> indexModels) {
        if (indexModels.isEmpty()) {
            return Mono.empty();
        }
        MongoCollection<T> collection = context.getCollection(clazz);
        Publisher<String> source = collection.createIndexes(indexModels);
        String collectionName = context.getEntity(clazz)
                .collectionName();
        return Flux.from(source)
                .then()
                .onErrorMap(t -> new RuntimeException(
                        "Failed to index the collection: \""
                                + collectionName
                                + "\"",
                        t))
                .doOnSuccess(ignored -> {
                    List<BsonDocument> indexDocs = indexModels.stream()
                            .map(indexModel -> indexModel.getKeys()
                                    .toBsonDocument())
                            .collect(CollectorUtil.toList(indexModels.size()));
                    String indexes = StringUtils.join(indexDocs, ", ");
                    LOGGER.info("Indexed the collection \"{}\": {}", collectionName, indexes);
                });
    }
apply the shard key to the filter so that mongos know to which MongoDB the command needs to
be sent for better performance
turms-server-common/src/main/java/im/turms/server/common/storage/mongo/operation/TurmsMongoOperations.java