SERVER-52809 Implement the new drop collection path in _shardsvrDropCollection

11 files changed, 556 insertions, 28 deletions

diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index cb48e076e87..e14b58c834f 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -337,10 +337,13 @@ env.Library(
         'move_chunk_command.cpp',
         'move_primary_command.cpp',
         'set_shard_version_command.cpp',
+        'drop_collection_coordinator.cpp',
+        'sharding_ddl_coordinator.cpp',
         'sharding_server_status.cpp',
         'sharding_state_command.cpp',
         'shardsvr_create_collection_command.cpp',
         'shardsvr_drop_collection_command.cpp',
+        'shardsvr_drop_collection_participant_command.cpp',
         'shardsvr_drop_database_command.cpp',
         'shardsvr_rename_collection.cpp',
         'shardsvr_refine_collection_shard_key_command.cpp',
diff --git a/src/mongo/db/s/dist_lock_manager.cpp b/src/mongo/db/s/dist_lock_manager.cpp
index 44e87291212..f58c54d8d56 100644
--- a/src/mongo/db/s/dist_lock_manager.cpp
+++ b/src/mongo/db/s/dist_lock_manager.cpp
@@ -73,8 +73,12 @@ DistLockManager::ScopedDistLock DistLockManager::ScopedDistLock::moveToAnotherTh
     return unownedScopedDistLock;
 }
 
+DistLockManager* DistLockManager::get(ServiceContext* service) {
+    return getDistLockManager(service).get();
+}
+
 DistLockManager* DistLockManager::get(OperationContext* opCtx) {
-    return getDistLockManager(opCtx->getServiceContext()).get();
+    return get(opCtx->getServiceContext());
 }
 
 void DistLockManager::create(ServiceContext* service,
diff --git a/src/mongo/db/s/dist_lock_manager.h b/src/mongo/db/s/dist_lock_manager.h
index dc2afed8cce..bffb49eaf7e 100644
--- a/src/mongo/db/s/dist_lock_manager.h
+++ b/src/mongo/db/s/dist_lock_manager.h
@@ -91,6 +91,7 @@ public:
     /**
      * Retrieves the DistLockManager singleton for the node.
      */
+    static DistLockManager* get(ServiceContext* service);
     static DistLockManager* get(OperationContext* opCtx);
     static void create(ServiceContext* service, std::unique_ptr<DistLockManager> distLockManager);
 
diff --git a/src/mongo/db/s/drop_collection_coordinator.cpp b/src/mongo/db/s/drop_collection_coordinator.cpp
new file mode 100644
index 00000000000..18220e4e0d7
--- /dev/null
+++ b/src/mongo/db/s/drop_collection_coordinator.cpp
@@ -0,0 +1,167 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+
+#include "mongo/db/s/drop_collection_coordinator.h"
+
+#include "mongo/db/api_parameters.h"
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/concurrency/lock_manager_defs.h"
+#include "mongo/db/s/database_sharding_state.h"
+#include "mongo/db/s/dist_lock_manager.h"
+#include "mongo/logv2/log.h"
+#include "mongo/s/catalog/sharding_catalog_client.h"
+#include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog/type_shard.h"
+#include "mongo/s/catalog/type_tags.h"
+#include "mongo/s/client/shard_registry.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/request_types/sharded_ddl_commands_gen.h"
+#include "mongo/util/assert_util.h"
+
+namespace mongo {
+
+static constexpr int kMaxNumStaleShardVersionRetries = 10;
+
+DropCollectionCoordinator::DropCollectionCoordinator(OperationContext* opCtx,
+                                                     const NamespaceString& nss)
+    : ShardingDDLCoordinator(nss), _serviceContext(opCtx->getServiceContext()) {
+    auto authSession = AuthorizationSession::get(opCtx->getClient());
+    _users =
+        userNameIteratorToContainer<std::vector<UserName>>(authSession->getImpersonatedUserNames());
+    _roles =
+        roleNameIteratorToContainer<std::vector<RoleName>>(authSession->getImpersonatedRoleNames());
+}
+
+void DropCollectionCoordinator::_sendDropCollToParticipants(OperationContext* opCtx) {
+    auto* const shardRegistry = Grid::get(opCtx)->shardRegistry();
+
+    const ShardsvrDropCollectionParticipant dropCollectionParticipant(_nss);
+
+    for (const auto& shardId : _participants) {
+        const auto& shard = uassertStatusOK(shardRegistry->getShard(opCtx, shardId));
+
+        const auto swDropResult = shard->runCommandWithFixedRetryAttempts(
+            opCtx,
+            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+            _nss.db().toString(),
+            CommandHelpers::appendMajorityWriteConcern(dropCollectionParticipant.toBSON({})),
+            Shard::RetryPolicy::kIdempotent);
+
+        uassertStatusOKWithContext(
+            Shard::CommandResponse::getEffectiveStatus(std::move(swDropResult)),
+            str::stream() << "Error dropping collection " << _nss.toString()
+                          << " on participant shard " << shardId);
+    }
+}
+
+void DropCollectionCoordinator::_removeCollMetadataFromConfig(OperationContext* opCtx) {
+    IgnoreAPIParametersBlock ignoreApiParametersBlock(opCtx);
+    const auto catalogClient = Grid::get(opCtx)->catalogClient();
+
+    ON_BLOCK_EXIT([this, opCtx] {
+        Grid::get(opCtx)->catalogCache()->invalidateCollectionEntry_LINEARIZABLE(_nss);
+    });
+
+    // Remove chunk data
+    uassertStatusOK(
+        catalogClient->removeConfigDocuments(opCtx,
+                                             ChunkType::ConfigNS,
+                                             BSON(ChunkType::ns(_nss.ns())),
+                                             ShardingCatalogClient::kMajorityWriteConcern));
+    // Remove tag data
+    uassertStatusOK(
+        catalogClient->removeConfigDocuments(opCtx,
+                                             TagsType::ConfigNS,
+                                             BSON(TagsType::ns(_nss.ns())),
+                                             ShardingCatalogClient::kMajorityWriteConcern));
+    // Remove coll metadata
+    uassertStatusOK(
+        catalogClient->removeConfigDocuments(opCtx,
+                                             CollectionType::ConfigNS,
+                                             BSON(CollectionType::kNssFieldName << _nss.ns()),
+                                             ShardingCatalogClient::kMajorityWriteConcern));
+}
+
+void DropCollectionCoordinator::_stopMigrations(OperationContext* opCtx) {
+    // TODO SERVER-53861 this will not stop current ongoing migrations
+    uassertStatusOK(Grid::get(opCtx)->catalogClient()->updateConfigDocument(
+        opCtx,
+        CollectionType::ConfigNS,
+        BSON(CollectionType::kNssFieldName << _nss.ns()),
+        BSON("$set" << BSON(CollectionType::kAllowMigrationsFieldName << false)),
+        false /* upsert */,
+        ShardingCatalogClient::kMajorityWriteConcern));
+}
+
+SemiFuture<void> DropCollectionCoordinator::runImpl(
+    std::shared_ptr<executor::TaskExecutor> executor) {
+    return ExecutorFuture<void>(executor, Status::OK())
+        .then([this, anchor = shared_from_this()]() {
+            ThreadClient tc{"DropCollectionCoordinator", _serviceContext};
+            auto opCtxHolder = tc->makeOperationContext();
+            auto* opCtx = opCtxHolder.get();
+
+            auto authSession = AuthorizationSession::get(opCtx->getClient());
+            authSession->setImpersonatedUserData(_users, _roles);
+
+            auto distLockManager = DistLockManager::get(_serviceContext);
+            const auto dbDistLock = uassertStatusOK(distLockManager->lock(
+                opCtx, _nss.db(), "DropCollection", DistLockManager::kDefaultLockTimeout));
+            const auto collDistLock = uassertStatusOK(distLockManager->lock(
+                opCtx, _nss.ns(), "DropCollection", DistLockManager::kDefaultLockTimeout));
+
+            _stopMigrations(opCtx);
+
+            const auto routingInfo = uassertStatusOK(
+                Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfoWithRefresh(opCtx, _nss));
+
+            _removeCollMetadataFromConfig(opCtx);
+
+            if (routingInfo.isSharded()) {
+                _participants = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx);
+            } else {
+                _participants = {routingInfo.dbPrimary()};
+            }
+
+            _sendDropCollToParticipants(opCtx);
+        })
+        .onError([this, anchor = shared_from_this()](const Status& status) {
+            LOGV2_ERROR(5280901,
+                        "Error running drop collection",
+                        "namespace"_attr = _nss,
+                        "error"_attr = redact(status));
+            return status;
+        })
+        .semi();
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/drop_collection_coordinator.h b/src/mongo/db/s/drop_collection_coordinator.h
new file mode 100644
index 00000000000..02347408d5b
--- /dev/null
+++ b/src/mongo/db/s/drop_collection_coordinator.h
@@ -0,0 +1,57 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/auth/user_name.h"
+#include "mongo/db/s/sharding_ddl_coordinator.h"
+#include "mongo/s/shard_id.h"
+
+namespace mongo {
+
+class DropCollectionCoordinator final
+    : public ShardingDDLCoordinator,
+      public std::enable_shared_from_this<DropCollectionCoordinator> {
+public:
+    DropCollectionCoordinator(OperationContext* opCtx, const NamespaceString& nss);
+
+private:
+    SemiFuture<void> runImpl(std::shared_ptr<executor::TaskExecutor> executor) override;
+
+    void _stopMigrations(OperationContext* opCtx);
+    void _removeCollMetadataFromConfig(OperationContext* opCtx);
+    void _sendDropCollToParticipants(OperationContext* opCtx);
+
+    ServiceContext* _serviceContext;
+    std::vector<ShardId> _participants;
+    std::vector<UserName> _users;
+    std::vector<RoleName> _roles;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/sharding_ddl_coordinator.cpp b/src/mongo/db/s/sharding_ddl_coordinator.cpp
new file mode 100644
index 00000000000..17c8842a01c
--- /dev/null
+++ b/src/mongo/db/s/sharding_ddl_coordinator.cpp
@@ -0,0 +1,71 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/s/sharding_ddl_coordinator.h"
+
+#include "mongo/db/s/database_sharding_state.h"
+#include "mongo/db/s/operation_sharding_state.h"
+#include "mongo/db/s/sharding_state.h"
+#include "mongo/s/grid.h"
+
+namespace mongo {
+
+ShardingDDLCoordinator::ShardingDDLCoordinator(const NamespaceString& ns) : _nss(ns){};
+
+SemiFuture<void> ShardingDDLCoordinator::run(OperationContext* opCtx) {
+    // Check that the operation context has a database version for this namespace
+    const auto clientDbVersion = OperationShardingState::get(opCtx).getDbVersion(_nss.db());
+    uassert(ErrorCodes::IllegalOperation,
+            str::stream() << "Request sent without attaching database version",
+            clientDbVersion);
+
+    // Checks that this is the primary shard for the namespace's db
+    const auto dbPrimaryShardId = [&]() {
+        Lock::DBLock dbWriteLock(opCtx, _nss.db(), MODE_IS);
+        auto dss = DatabaseShardingState::get(opCtx, _nss.db());
+        auto dssLock = DatabaseShardingState::DSSLock::lockShared(opCtx, dss);
+        // The following call will also ensure that the database version matches
+        return dss->getDatabaseInfo(opCtx, dssLock).getPrimary();
+    }();
+
+    const auto thisShardId = ShardingState::get(opCtx)->shardId();
+
+    uassert(ErrorCodes::IllegalOperation,
+            str::stream() << "This is not the primary shard for db " << _nss.db()
+                          << " expected: " << dbPrimaryShardId << " shardId: " << thisShardId,
+            dbPrimaryShardId == thisShardId);
+
+    return runImpl(Grid::get(opCtx)->getExecutorPool()->getFixedExecutor());
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/sharding_ddl_coordinator.h b/src/mongo/db/s/sharding_ddl_coordinator.h
new file mode 100644
index 00000000000..e6575d247e7
--- /dev/null
+++ b/src/mongo/db/s/sharding_ddl_coordinator.h
@@ -0,0 +1,51 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/executor/task_executor.h"
+#include "mongo/util/future.h"
+
+namespace mongo {
+
+class ShardingDDLCoordinator {
+public:
+    ShardingDDLCoordinator(const NamespaceString& nss);
+    SemiFuture<void> run(OperationContext* opCtx);
+
+protected:
+    NamespaceString _nss;
+
+private:
+    virtual SemiFuture<void> runImpl(std::shared_ptr<executor::TaskExecutor>) = 0;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/s/shardsvr_drop_collection_command.cpp b/src/mongo/db/s/shardsvr_drop_collection_command.cpp
index 1fe52939abb..0cd8bd76f20 100644
--- a/src/mongo/db/s/shardsvr_drop_collection_command.cpp
+++ b/src/mongo/db/s/shardsvr_drop_collection_command.cpp
@@ -29,18 +29,35 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/commands.h"
-#include "mongo/db/drop_database_gen.h"
-#include "mongo/db/s/sharding_logging.h"
-#include "mongo/s/catalog/type_database.h"
-#include "mongo/s/catalog_cache.h"
+#include "mongo/db/curop.h"
+#include "mongo/db/s/drop_collection_coordinator.h"
+#include "mongo/db/s/sharding_state.h"
+#include "mongo/logv2/log.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/request_types/sharded_ddl_commands_gen.h"
+#include "mongo/s/sharded_collections_ddl_parameters_gen.h"
 
 namespace mongo {
 namespace {
 
+void dropCollectionLegacy(OperationContext* opCtx, const NamespaceString& nss) {
+    auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
+    const auto cmdResponse = configShard->runCommandWithFixedRetryAttempts(
+        opCtx,
+        ReadPreferenceSetting(ReadPreference::PrimaryOnly),
+        "admin",
+        CommandHelpers::appendMajorityWriteConcern(
+            BSON("_configsvrDropCollection" << nss.toString()), opCtx->getWriteConcern()),
+        Shard::RetryPolicy::kIdempotent);
+
+    uassertStatusOK(Shard::CommandResponse::getEffectiveStatus(cmdResponse));
+}
+
 class ShardsvrDropCollectionCommand final : public TypedCommand<ShardsvrDropCollectionCommand> {
 public:
     bool acceptsAnyApiVersionParameters() const override {
@@ -63,33 +80,57 @@ public:
         using InvocationBase::InvocationBase;
 
         void typedRun(OperationContext* opCtx) {
-            uassert(ErrorCodes::IllegalOperation,
-                    "_shardsvrDropCollection can only be run on primary shard servers",
-                    serverGlobalParams.clusterRole == ClusterRole::ShardServer);
-
-            auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-            auto cmdResponse = uassertStatusOK(configShard->runCommandWithFixedRetryAttempts(
-                opCtx,
-                ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                "admin",
-                CommandHelpers::appendMajorityWriteConcern(
-                    BSON("_configsvrDropCollection" << ns().toString()), opCtx->getWriteConcern()),
-                Shard::RetryPolicy::kIdempotent));
-
-            uassertStatusOK(cmdResponse.commandStatus);
+            uassertStatusOK(ShardingState::get(opCtx)->canAcceptShardedCommands());
+
+            uassert(ErrorCodes::InvalidOptions,
+                    str::stream() << Request::kCommandName
+                                  << " must be called with majority writeConcern, got "
+                                  << opCtx->getWriteConcern().wMode,
+                    opCtx->getWriteConcern().wMode == WriteConcernOptions::kMajority);
+
+            if (!feature_flags::gShardingFullDDLSupport.isEnabled(
+                    serverGlobalParams.featureCompatibility) ||
+                feature_flags::gDisableIncompleteShardingDDLSupport.isEnabled(
+                    serverGlobalParams.featureCompatibility)) {
+                LOGV2_DEBUG(5280951,
+                            1,
+                            "Running legacy drop collection procedure",
+                            "namespace"_attr = ns());
+                dropCollectionLegacy(opCtx, ns());
+                return;
+            }
+
+            LOGV2_DEBUG(
+                5280952, 1, "Running new drop collection procedure", "namespace"_attr = ns());
+
+            // Since this operation is not directly writing locally we need to force its db
+            // profile level increase in order to be logged in "<db>.system.profile"
+            CurOp::get(opCtx)->raiseDbProfileLevel(
+                CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(ns().db()));
+
+            auto dropCollectionCoordinator =
+                std::make_shared<DropCollectionCoordinator>(opCtx, ns());
+            dropCollectionCoordinator->run(opCtx).get();
+        }
+
+    private:
+        NamespaceString ns() const override {
+            return request().getNamespace();
         }
 
         bool supportsWriteConcern() const override {
             return true;
         }
 
-        void doCheckAuthorization(OperationContext*) const override {}
-
-        NamespaceString ns() const override {
-            return request().getNamespace();
+        void doCheckAuthorization(OperationContext* opCtx) const override {
+            uassert(ErrorCodes::Unauthorized,
+                    "Unauthorized",
+                    AuthorizationSession::get(opCtx->getClient())
+                        ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                           ActionType::internal));
         }
     };
 } sharsvrdDropCollectionCommand;
 
 }  // namespace
-}  // namespace mongo
\ No newline at end of file
+}  // namespace mongo
diff --git a/src/mongo/db/s/shardsvr_drop_collection_participant_command.cpp b/src/mongo/db/s/shardsvr_drop_collection_participant_command.cpp
new file mode 100644
index 00000000000..5de98d41182
--- /dev/null
+++ b/src/mongo/db/s/shardsvr_drop_collection_participant_command.cpp
@@ -0,0 +1,122 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/drop_collection.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/commands.h"
+#include "mongo/db/s/collection_sharding_runtime.h"
+#include "mongo/db/s/sharding_state.h"
+#include "mongo/logv2/log.h"
+#include "mongo/s/request_types/sharded_ddl_commands_gen.h"
+
+namespace mongo {
+namespace {
+
+class ShardsvrDropCollectionParticipantCommand final
+    : public TypedCommand<ShardsvrDropCollectionParticipantCommand> {
+public:
+    bool acceptsAnyApiVersionParameters() const override {
+        return true;
+    }
+
+    AllowedOnSecondary secondaryAllowed(ServiceContext*) const override {
+        return Command::AllowedOnSecondary::kNever;
+    }
+
+    std::string help() const override {
+        return "Internal command, which is exported by secondary sharding servers. Do not call "
+               "directly. Participates in droping a collection.";
+    }
+
+    using Request = ShardsvrDropCollectionParticipant;
+
+    class Invocation final : public InvocationBase {
+    public:
+        using InvocationBase::InvocationBase;
+
+        void typedRun(OperationContext* opCtx) {
+            uassertStatusOK(ShardingState::get(opCtx)->canAcceptShardedCommands());
+            uassert(ErrorCodes::InvalidOptions,
+                    str::stream() << Request::kCommandName
+                                  << " must be called with majority writeConcern, got "
+                                  << opCtx->getWriteConcern().wMode,
+                    opCtx->getWriteConcern().wMode == WriteConcernOptions::kMajority);
+
+            DropReply result;
+            try {
+
+                uassertStatusOK(dropCollection(
+                    opCtx,
+                    ns(),
+                    &result,
+                    DropCollectionSystemCollectionMode::kDisallowSystemCollectionDrops));
+
+            } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
+                LOGV2_DEBUG(5280920,
+                            1,
+                            "Namespace not found while trying to delete local collection",
+                            "namespace"_attr = ns());
+            }
+
+            {
+                // Clear CollectionShardingRuntime entry
+                UninterruptibleLockGuard noInterrupt(opCtx->lockState());
+                Lock::DBLock dbLock(opCtx, ns().db(), MODE_IX);
+                Lock::CollectionLock collLock(opCtx, ns(), MODE_IX);
+                auto* csr = CollectionShardingRuntime::get(opCtx, ns());
+                csr->clearFilteringMetadata(opCtx);
+            }
+        }
+
+    private:
+        NamespaceString ns() const override {
+            return request().getNamespace();
+        }
+
+        bool supportsWriteConcern() const override {
+            return true;
+        }
+
+        void doCheckAuthorization(OperationContext* opCtx) const override {
+            uassert(ErrorCodes::Unauthorized,
+                    "Unauthorized",
+                    AuthorizationSession::get(opCtx->getClient())
+                        ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                           ActionType::internal));
+        }
+    };
+} sharsvrdDropCollectionParticipantCommand;
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/request_types/sharded_ddl_commands.idl b/src/mongo/s/request_types/sharded_ddl_commands.idl
index 85a494641ea..56a35a942e0 100644
--- a/src/mongo/s/request_types/sharded_ddl_commands.idl
+++ b/src/mongo/s/request_types/sharded_ddl_commands.idl
@@ -106,8 +106,14 @@ commands:
         command_name: _shardsvrDropCollection
         namespace: concatenate_with_db
         cpp_name: ShardsvrDropCollection
-        strict: true
-        reply_type: DropDatabaseReply
+        strict: false
+
+    _shardsvrDropCollectionParticipant:
+        description: "Parser for the _shardsvrDropCollectionParticipant command"
+        command_name: _shardsvrDropCollectionParticipant
+        namespace: concatenate_with_db
+        cpp_name: ShardsvrDropCollectionParticipant
+        strict: false
 
     _shardsvrRenameCollection:
         command_name: _shardsvrRenameCollection
diff --git a/src/mongo/s/sharded_collections_ddl_parameters.idl b/src/mongo/s/sharded_collections_ddl_parameters.idl
index f7d798332db..18b5b8251cb 100644
--- a/src/mongo/s/sharded_collections_ddl_parameters.idl
+++ b/src/mongo/s/sharded_collections_ddl_parameters.idl
@@ -31,11 +31,16 @@ global:
   cpp_namespace: "mongo::feature_flags"
 
 feature_flags:
-  shardingFullDDLSupport:
+  featureFlagShardingFullDDLSupport:
     description: "Ensures extra guarantees on DDL operations under a sharded cluster."
     cpp_varname: gShardingFullDDLSupport
     default: false
-  
+
+  featureFlagDisableIncompleteShardingDDLSupport:
+    description: "Disable incomplete sharding DDL features."
+    cpp_varname: gDisableIncompleteShardingDDLSupport
+    default: false
+
   shardingFullDDLSupportDistLocksOnStepDown:
     description: "Once implemented allow usage of Distributed locks on shards."
     cpp_varname: gShardingFullDDLSupportDistLocksOnStepDown