From ce33366396467ca88339e24a59ed808ebcb36e50 Mon Sep 17 00:00:00 2001
From: Xiaoxi Wang <xiaoxi.wang@snowflake.com>
Date: Tue, 5 Apr 2022 17:05:51 -0700
Subject: [PATCH] only add mutations can change configuration (#6760)

---
 fdbclient/DatabaseConfiguration.cpp | 5 +++++
 fdbclient/DatabaseConfiguration.h   | 2 ++
 fdbserver/ClusterRecovery.actor.cpp | 3 ++-
 3 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/fdbclient/DatabaseConfiguration.cpp b/fdbclient/DatabaseConfiguration.cpp
index 7978c14fbb..657129f86e 100644
--- a/fdbclient/DatabaseConfiguration.cpp
+++ b/fdbclient/DatabaseConfiguration.cpp
@@ -660,6 +660,11 @@ void DatabaseConfiguration::applyMutation(MutationRef m) {
 	}
 }
 
+bool DatabaseConfiguration::involveMutation(MutationRef m) {
+	return (m.type == MutationRef::SetValue && m.param1.startsWith(configKeysPrefix)) ||
+	       (m.type == MutationRef::ClearRange && KeyRangeRef(m.param1, m.param2).intersects(configKeys));
+}
+
 bool DatabaseConfiguration::set(KeyRef key, ValueRef value) {
 	makeConfigurationMutable();
 	mutableConfiguration.get()[key.toString()] = value.toString();
diff --git a/fdbclient/DatabaseConfiguration.h b/fdbclient/DatabaseConfiguration.h
index 192d399bc2..363b48e4b6 100644
--- a/fdbclient/DatabaseConfiguration.h
+++ b/fdbclient/DatabaseConfiguration.h
@@ -104,6 +104,8 @@ struct DatabaseConfiguration {
 	DatabaseConfiguration();
 
 	void applyMutation(MutationRef mutation);
+	// return true if mutation will cause configuration changes
+	bool involveMutation(MutationRef mutation);
 	bool set(KeyRef key,
 	         ValueRef value); // Returns true if a configuration option that requires recovery to take effect is changed
 	bool clear(KeyRangeRef keys);
diff --git a/fdbserver/ClusterRecovery.actor.cpp b/fdbserver/ClusterRecovery.actor.cpp
index 180673f850..fd8bbbd0c7 100644
--- a/fdbserver/ClusterRecovery.actor.cpp
+++ b/fdbserver/ClusterRecovery.actor.cpp
@@ -869,7 +869,8 @@ ACTOR Future<Standalone<CommitTransactionRef>> provisionalMaster(Reference<Clust
 					    .detail("MType", m->type)
 					    .detail("Param1", m->param1)
 					    .detail("Param2", m->param2);
-					if (isMetadataMutation(*m)) {
+					// emergency transaction only mean to do configuration change
+					if (parent->configuration.involveMutation(*m)) {
 						// We keep the mutations and write conflict ranges from this transaction, but not its read
 						// conflict ranges
 						Standalone<CommitTransactionRef> out;