Permalink
Browse files

SERVER-27149 Don't sync from nodes in an older term.

  • Loading branch information...
stbrody committed Nov 21, 2016
1 parent 524cc65 commit 138402742f13b1cf85b021966eb27f2e33667cca
@@ -1223,8 +1223,7 @@ void DataReplicator::_setState_inlock(const DataReplicatorState& newState) {
}
StatusWith<HostAndPort> DataReplicator::_chooseSyncSource_inlock() {
- auto syncSource =
- _opts.syncSourceSelector->chooseNewSyncSource(_lastFetched.opTime.getTimestamp());
+ auto syncSource = _opts.syncSourceSelector->chooseNewSyncSource(_lastFetched.opTime);
if (syncSource.empty()) {
return Status{ErrorCodes::InvalidSyncSource,
str::stream() << "No valid sync source available. Our last fetched optime: "
@@ -85,7 +85,7 @@ class SyncSourceSelectorMock : public SyncSourceSelector {
public:
SyncSourceSelectorMock(const HostAndPort& syncSource) : _syncSource(syncSource) {}
void clearSyncSourceBlacklist() override {}
- HostAndPort chooseNewSyncSource(const Timestamp& ts) override {
+ HostAndPort chooseNewSyncSource(const OpTime& ot) override {
HostAndPort result = _syncSource;
return result;
}
@@ -120,8 +120,8 @@ class DataReplicatorTest : public executor::ThreadPoolExecutorTest, public SyncS
void clearSyncSourceBlacklist() override {
_syncSourceSelector->clearSyncSourceBlacklist();
}
- HostAndPort chooseNewSyncSource(const Timestamp& ts) override {
- return _syncSourceSelector->chooseNewSyncSource(ts);
+ HostAndPort chooseNewSyncSource(const OpTime& ot) override {
+ return _syncSourceSelector->chooseNewSyncSource(ot);
}
void blacklistSyncSource(const HostAndPort& host, Date_t until) override {
_syncSourceSelector->blacklistSyncSource(host, until);
@@ -164,7 +164,7 @@ void OplogReader::connectToSyncSource(OperationContext* txn,
invariant(conn() == NULL);
while (true) {
- HostAndPort candidate = replCoord->chooseNewSyncSource(lastOpTimeFetched.getTimestamp());
+ HostAndPort candidate = replCoord->chooseNewSyncSource(lastOpTimeFetched);
if (candidate.empty()) {
if (oldestOpTimeSeen == sentinel) {
@@ -3037,15 +3037,15 @@ bool ReplicationCoordinatorImpl::isReplEnabled() const {
return getReplicationMode() != modeNone;
}
-HostAndPort ReplicationCoordinatorImpl::chooseNewSyncSource(const Timestamp& lastTimestampFetched) {
+HostAndPort ReplicationCoordinatorImpl::chooseNewSyncSource(const OpTime& lastOpTimeFetched) {
LockGuard topoLock(_topoMutex);
HostAndPort oldSyncSource = _topCoord->getSyncSourceAddress();
auto chainingPreference = isCatchingUp()
? TopologyCoordinator::ChainingPreference::kAllowChaining
: TopologyCoordinator::ChainingPreference::kUseConfiguration;
- HostAndPort newSyncSource = _topCoord->chooseNewSyncSource(
- _replExecutor.now(), lastTimestampFetched, chainingPreference);
+ HostAndPort newSyncSource =
+ _topCoord->chooseNewSyncSource(_replExecutor.now(), lastOpTimeFetched, chainingPreference);
stdx::lock_guard<stdx::mutex> lock(_mutex);
// If we lost our sync source, schedule new heartbeats immediately to update our knowledge
@@ -261,7 +261,7 @@ class ReplicationCoordinatorImpl : public ReplicationCoordinator {
virtual bool isReplEnabled() const override;
- virtual HostAndPort chooseNewSyncSource(const Timestamp& lastTimestampFetched) override;
+ virtual HostAndPort chooseNewSyncSource(const OpTime& lastOpTimeFetched) override;
virtual void blacklistSyncSource(const HostAndPort& host, Date_t until) override;
@@ -370,7 +370,7 @@ Status ReplicationCoordinatorMock::checkReplEnabledForCommand(BSONObjBuilder* re
return Status::OK();
}
-HostAndPort ReplicationCoordinatorMock::chooseNewSyncSource(const Timestamp& lastTimestampFetched) {
+HostAndPort ReplicationCoordinatorMock::chooseNewSyncSource(const OpTime& lastOpTimeFetched) {
return HostAndPort();
}
@@ -205,7 +205,7 @@ class ReplicationCoordinatorMock : public ReplicationCoordinator {
virtual Status checkReplEnabledForCommand(BSONObjBuilder* result);
- virtual HostAndPort chooseNewSyncSource(const Timestamp& lastTimestampFetched);
+ virtual HostAndPort chooseNewSyncSource(const OpTime& lastOpTimeFetched);
virtual void blacklistSyncSource(const HostAndPort& host, Date_t until);
@@ -144,7 +144,7 @@ bool SyncSourceResolver::_isShuttingDown() const {
StatusWith<HostAndPort> SyncSourceResolver::_chooseNewSyncSource() {
HostAndPort candidate;
try {
- candidate = _syncSourceSelector->chooseNewSyncSource(_lastOpTimeFetched.getTimestamp());
+ candidate = _syncSourceSelector->chooseNewSyncSource(_lastOpTimeFetched);
} catch (...) {
return exceptionToStatus();
}
@@ -70,9 +70,9 @@ class TaskExecutorWithFailureInScheduleRemoteCommand : public unittest::TaskExec
class SyncSourceSelectorMock : public SyncSourceSelector {
public:
void clearSyncSourceBlacklist() override {}
- HostAndPort chooseNewSyncSource(const Timestamp& ts) override {
+ HostAndPort chooseNewSyncSource(const OpTime& ot) override {
chooseNewSyncSourceHook();
- lastTimestampFetched = ts;
+ lastOpTimeFetched = ot;
return syncSource;
}
void blacklistSyncSource(const HostAndPort& host, Date_t until) override {
@@ -84,7 +84,7 @@ class SyncSourceSelectorMock : public SyncSourceSelector {
}
HostAndPort syncSource = HostAndPort("host1", 1234);
- Timestamp lastTimestampFetched;
+ OpTime lastOpTimeFetched;
stdx::function<void()> chooseNewSyncSourceHook = []() {};
HostAndPort blacklistHost;
@@ -264,7 +264,7 @@ TEST_F(SyncSourceResolverTest,
// Resolver invokes callback with empty host and becomes inactive immediately.
ASSERT_FALSE(_resolver->isActive());
ASSERT_EQUALS(HostAndPort(), unittest::assertGet(_response.syncSourceStatus));
- ASSERT_EQUALS(lastOpTimeFetched.getTimestamp(), _selector->lastTimestampFetched);
+ ASSERT_EQUALS(lastOpTimeFetched, _selector->lastOpTimeFetched);
// Cannot restart a completed resolver.
ASSERT_EQUALS(ErrorCodes::ShutdownInProgress, _resolver->startup());
@@ -64,7 +64,7 @@ class SyncSourceSelector {
/**
* Chooses a viable sync source, or, if none available, returns empty HostAndPort.
*/
- virtual HostAndPort chooseNewSyncSource(const Timestamp& lastTimestampFetched) = 0;
+ virtual HostAndPort chooseNewSyncSource(const OpTime& lastOpTimeFetched) = 0;
/**
* Blacklists choosing 'host' as a sync source until time 'until'.
@@ -138,7 +138,7 @@ class TopologyCoordinator {
* Chooses and sets a new sync source, based on our current knowledge of the world.
*/
virtual HostAndPort chooseNewSyncSource(Date_t now,
- const Timestamp& lastTimestampFetched,
+ const OpTime& lastOpTimeFetched,
ChainingPreference chainingPreference) = 0;
/**
@@ -154,7 +154,7 @@ HostAndPort TopologyCoordinatorImpl::getSyncSourceAddress() const {
}
HostAndPort TopologyCoordinatorImpl::chooseNewSyncSource(Date_t now,
- const Timestamp& lastTimestampFetched,
+ const OpTime& lastOpTimeFetched,
ChainingPreference chainingPreference) {
// If we are not a member of the current replica set configuration, no sync source is valid.
if (_selfIndex == -1) {
@@ -305,12 +305,12 @@ HostAndPort TopologyCoordinatorImpl::chooseNewSyncSource(Date_t now,
}
}
// only consider candidates that are ahead of where we are
- if (it->getAppliedOpTime().getTimestamp() <= lastTimestampFetched) {
+ if (it->getAppliedOpTime() <= lastOpTimeFetched) {
LOG(1) << "Cannot select sync source equal to or behind our last fetched optime. "
- << "My last fetched oplog timestamp: " << lastTimestampFetched.toBSON()
- << ", latest oplog timestamp of sync candidate "
+ << "My last fetched oplog optime: " << lastOpTimeFetched.toBSON()
+ << ", latest oplog optime of sync candidate "
<< itMemberConfig.getHostAndPort() << ": "
- << it->getAppliedOpTime().getTimestamp().toBSON();
+ << it->getAppliedOpTime().toBSON();
continue;
}
// Candidate cannot be more latent than anything we've already considered.
@@ -151,8 +151,8 @@ class TopologyCoordinatorImpl : public TopologyCoordinator {
virtual UpdateTermResult updateTerm(long long term, Date_t now);
virtual void setForceSyncSourceIndex(int index);
virtual HostAndPort chooseNewSyncSource(Date_t now,
- const Timestamp& lastTimestampFetched,
- ChainingPreference chainingPreference);
+ const OpTime& lastOpTimeFetched,
+ ChainingPreference chainingPreference) override;
virtual void blacklistSyncSource(const HostAndPort& host, Date_t until);
virtual void unblacklistSyncSource(const HostAndPort& host, Date_t now);
virtual void clearSyncSourceBlacklist();
Oops, something went wrong.

0 comments on commit 1384027

Please sign in to comment.