diff --git a/fdbserver/datadistributor/DataDistribution.cpp b/fdbserver/datadistributor/DataDistribution.cpp index 81047562e15..35c4057cd8f 100644 --- a/fdbserver/datadistributor/DataDistribution.cpp +++ b/fdbserver/datadistributor/DataDistribution.cpp @@ -2914,6 +2914,41 @@ Future bulkDumpCore(Reference self, Future readyToS } } +// These actors read or write system keys through a real Database and are not part of MockDD's transaction +// processor contract. +void addProductionOnlyDataDistributionActors(Reference self, std::vector>& actors) { + if (bulkLoadIsEnabled(self->initData->bulkLoadMode)) { + TraceEvent(SevInfo, "DDBulkLoadModeEnabled", self->ddId) + .detail("UsableRegions", self->configuration.usableRegions); + self->bulkLoadEnabled = true; + if (self->configuration.usableRegions > 1) { + // The core actor to handle bulkload tasks + actors.push_back(bulkLoadTaskCore(self, self->initialized.getFuture() && remoteRecovered(self->dbInfo))); + // The core actor to convert a bulkload job to tasks which are executed by bulkLoadTaskCore + actors.push_back(bulkLoadJobCore(self, self->initialized.getFuture() && remoteRecovered(self->dbInfo))); + } else { + actors.push_back(bulkLoadTaskCore(self, self->initialized.getFuture())); + actors.push_back(bulkLoadJobCore(self, self->initialized.getFuture())); + } + } else { + self->bulkLoadTaskCollection->removeBulkLoadJobRange(); + // Monitor for bulkLoadMode changes and spawn actors dynamically. + // This is critical for restore operations which set bulkLoadMode=1 after DD starts. + actors.push_back(monitorBulkLoadModeAndSpawnActors(self, self->initialized.getFuture())); + } + + // Always spawn bulkDumpCore for production DD - it will dynamically check the mode + // NOTE: BulkDump does NOT require remoteRecovered() in HA configurations. + // BulkDump is read-only: it reads from primary DC and writes to external storage (S3). + // Waiting for remoteRecovered() caused hangs when remote DC couldn't form teams. + TraceEvent(SevInfo, "DDBulkDumpCoreSpawned", self->ddId) + .detail("UsableRegions", self->configuration.usableRegions) + .detail("InitialMode", self->initData->bulkDumpMode); + actors.push_back(bulkDumpCore(self, self->initialized.getFuture())); + + actors.push_back(periodicAuditLocationMetadata(self)); +} + // Runs the data distribution algorithm for FDB, including the DD Queue, DD tracker, and DD team collection Future dataDistribution(Reference self, PromiseStream getShardMetricsList, @@ -2993,7 +3028,9 @@ Future dataDistribution(Reference self, } actors.push_back(self->pollMoveKeysLock()); - actors.push_back(monitorBackupPartitionRequired(self->txnProcessor->context(), &shards, self->ddId)); + if (!isMocked) { + actors.push_back(monitorBackupPartitionRequired(self->txnProcessor->context(), &shards, self->ddId)); + } self->context->tracker = makeReference( DataDistributionTrackerInitParams{ .db = self->txnProcessor, @@ -3120,39 +3157,12 @@ Future dataDistribution(Reference self, actors.push_back(monitorPhysicalShardStatus(self->physicalShardCollection)); } - if (bulkLoadIsEnabled(self->initData->bulkLoadMode)) { - TraceEvent(SevInfo, "DDBulkLoadModeEnabled", self->ddId) - .detail("UsableRegions", self->configuration.usableRegions); - self->bulkLoadEnabled = true; - if (self->configuration.usableRegions > 1) { - // The core actor to handle bulkload tasks - actors.push_back( - bulkLoadTaskCore(self, self->initialized.getFuture() && remoteRecovered(self->dbInfo))); - // The core actor to convert a bulkload job to tasks which are executed by bulkLoadTaskCore - actors.push_back( - bulkLoadJobCore(self, self->initialized.getFuture() && remoteRecovered(self->dbInfo))); - } else { - actors.push_back(bulkLoadTaskCore(self, self->initialized.getFuture())); - actors.push_back(bulkLoadJobCore(self, self->initialized.getFuture())); - } - } else { + if (isMocked) { self->bulkLoadTaskCollection->removeBulkLoadJobRange(); - // Monitor for bulkLoadMode changes and spawn actors dynamically. - // This is critical for restore operations which set bulkLoadMode=1 after DD starts. - actors.push_back(monitorBulkLoadModeAndSpawnActors(self, self->initialized.getFuture())); + } else { + addProductionOnlyDataDistributionActors(self, actors); } - // Always spawn bulkDumpCore - it will dynamically check the mode - // NOTE: BulkDump does NOT require remoteRecovered() in HA configurations. - // BulkDump is read-only: it reads from primary DC and writes to external storage (S3). - // Waiting for remoteRecovered() caused hangs when remote DC couldn't form teams. - TraceEvent(SevInfo, "DDBulkDumpCoreSpawned", self->ddId) - .detail("UsableRegions", self->configuration.usableRegions) - .detail("InitialMode", self->initData->bulkDumpMode); - actors.push_back(bulkDumpCore(self, self->initialized.getFuture())); - - actors.push_back(periodicAuditLocationMetadata(self)); - actors.push_back(monitorShardEncodeKnob(self->ddId)); co_await waitForAll(actors); diff --git a/tests/fast/MockDDReadWrite.toml b/tests/fast/MockDDReadWrite.toml index 7eb334e82ed..9d953fef24f 100644 --- a/tests/fast/MockDDReadWrite.toml +++ b/tests/fast/MockDDReadWrite.toml @@ -3,7 +3,7 @@ testClass = 'MockDD' [[knobs]] enable_dd_physical_shard = false -storage_quota_enabled = false +shard_encode_location_metadata = false [[test]] testTitle = 'MockDDTracker' @@ -24,4 +24,4 @@ useDB = false testName = 'MockDDReadWrite' keySpaceCount = 1000 testDuration = 500.0 - simpleConfig = true \ No newline at end of file + simpleConfig = true