@@ -1712,7 +1712,7 @@ ACTOR static Future<Void> startMoveShards(Database occ,
17121712 serverListEntries.push_back (tr.get (serverListKeyFor (servers[s])));
17131713 }
17141714 std::vector<Optional<Value>> serverListValues = wait (getAll (serverListEntries));
1715-
1715+ state std::unordered_map<std::string, std::vector<std::string>> dcServers;
17161716 for (int s = 0 ; s < serverListValues.size (); s++) {
17171717 if (!serverListValues[s].present ()) {
17181718 // Attempt to move onto a server that isn't in serverList (removed or never added to the
@@ -1721,6 +1721,13 @@ ACTOR static Future<Void> startMoveShards(Database occ,
17211721 // TODO(psm): Mark the data move as 'deleting'.
17221722 throw move_to_removed_server ();
17231723 }
1724+ auto si = decodeServerListValue (serverListValues[s].get ());
1725+ ASSERT (si.id () == servers[s]);
1726+ auto it = dcServers.find (si.locality .describeDcId ());
1727+ if (it == dcServers.end ()) {
1728+ dcServers[si.locality .describeDcId ()] = std::vector<std::string>();
1729+ }
1730+ dcServers[si.locality .describeDcId ()].push_back (si.id ().shortString ());
17241731 }
17251732
17261733 currentKeys = KeyRangeRef (begin, keys.end );
@@ -1733,6 +1740,15 @@ ACTOR static Future<Void> startMoveShards(Database occ,
17331740 state Key endKey = old.back ().key ;
17341741 currentKeys = KeyRangeRef (currentKeys.begin , endKey);
17351742
1743+ if (ranges.front () != currentKeys) {
1744+ TraceEvent (" MoveShardsPartialRange" )
1745+ .detail (" ExpectedRange" , ranges.front ())
1746+ .detail (" ActualRange" , currentKeys)
1747+ .detail (" DataMoveId" , dataMoveId)
1748+ .detail (" RowLimit" , SERVER_KNOBS->MOVE_SHARD_KRM_ROW_LIMIT )
1749+ .detail (" ByteLimit" , SERVER_KNOBS->MOVE_SHARD_KRM_BYTE_LIMIT );
1750+ }
1751+
17361752 // Check that enough servers for each shard are in the correct state
17371753 state RangeResult UIDtoTagMap = wait (tr.getRange (serverTagKeys, CLIENT_KNOBS->TOO_MANY ));
17381754 ASSERT (!UIDtoTagMap.more && UIDtoTagMap.size () < CLIENT_KNOBS->TOO_MANY );
@@ -1806,6 +1822,7 @@ ACTOR static Future<Void> startMoveShards(Database occ,
18061822 TraceEvent (
18071823 SevWarn, " StartMoveShardsCancelConflictingDataMove" , relocationIntervalId)
18081824 .detail (" Range" , rangeIntersectKeys)
1825+ .detail (" CurrentDataMoveRange" , ranges[0 ])
18091826 .detail (" DataMoveID" , dataMoveId.toString ())
18101827 .detail (" ExistingDataMoveID" , destId.toString ());
18111828 wait (cleanUpDataMove (occ, destId, lock, startMoveKeysLock, keys, ddEnabledState));
@@ -1868,6 +1885,20 @@ ACTOR static Future<Void> startMoveShards(Database occ,
18681885 dataMove.ranges .clear ();
18691886 dataMove.ranges .push_back (KeyRangeRef (keys.begin , currentKeys.end ));
18701887 dataMove.dest .insert (servers.begin (), servers.end ());
1888+ dataMove.dcTeamIds = std::unordered_map<std::string, std::string>();
1889+ for (auto & [dc, serverIds] : dcServers) {
1890+ std::sort (serverIds.begin (), serverIds.end ());
1891+ std::string teamId;
1892+ for (const auto & serverId : serverIds) {
1893+ if (teamId.size () == 0 ) {
1894+ teamId = serverId;
1895+ } else {
1896+ teamId += " ," + serverId;
1897+ }
1898+ }
1899+ // Use the concatenated server ids as the team id to avoid conflicts.
1900+ dataMove.dcTeamIds .get ()[dc] = teamId;
1901+ }
18711902 }
18721903
18731904 if (currentKeys.end == keys.end ) {
0 commit comments