Fix wrongly try to rewrite the namespace in the cluster mode

src/server/namespace.cc

@@ -53,6 +53,9 @@ bool Namespace::IsAllowModify() const {
 
 Status Namespace::LoadAndRewrite() {
   auto config = storage_->GetConfig();
+  // Namespace is NOT allowed in the cluster mode, so we don't need to rewrite here.
+  if (config->cluster_enabled) return Status::OK();
+
   // Load from the configuration file first
   tokens_ = config->load_tokens;
 

src/storage/redis_pubsub.cc

@@ -23,6 +23,9 @@
 namespace redis {
 
 rocksdb::Status PubSub::Publish(const Slice &channel, const Slice &value) {
+  if (storage_->GetConfig()->IsSlave()) {
+    return rocksdb::Status::NotSupported("can't publish to db in slave mode");
+  }
   auto batch = storage_->GetWriteBatchBase();
   batch->Put(pubsub_cf_handle_, channel, value);
   return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());

src/storage/storage.cc

@@ -863,6 +863,9 @@ ObserverOrUniquePtr<rocksdb::WriteBatchBase> Storage::GetWriteBatchBase() {
 }
 
 Status Storage::WriteToPropagateCF(const std::string &key, const std::string &value) {
+  if (config_->IsSlave()) {
+    return {Status::NotOK, "cannot write to propagate column family in slave mode"};
+  }
   auto batch = GetWriteBatchBase();
   auto cf = GetCFHandle(kPropagateColumnFamilyName);
   batch->Put(cf, key, value);

tests/gocase/integration/replication/replication_test.go

@@ -32,6 +32,64 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestClusterReplication(t *testing.T) {
+	ctx := context.Background()
+
+	masterSrv := util.StartServer(t, map[string]string{"cluster-enabled": "yes"})
+	defer func() { masterSrv.Close() }()
+	masterClient := masterSrv.NewClient()
+	defer func() { require.NoError(t, masterClient.Close()) }()
+	masterNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+	require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODEID", masterNodeID).Err())
+
+	replicaSrv := util.StartServer(t, map[string]string{
+		"cluster-enabled": "yes",
+		// enabled the replication namespace to reproduce the issue #2214
+		"repl-namespace-enabled": "yes",
+	})
+	defer func() { replicaSrv.Close() }()
+	replicaClient := replicaSrv.NewClient()
+	// allow to run the read-only command in the replica
+	require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+	defer func() { require.NoError(t, replicaClient.Close()) }()
+	replicaNodeID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+	require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODEID", replicaNodeID).Err())
+
+	clusterNodes := fmt.Sprintf("%s 127.0.0.1 %d master - 0-16383", masterNodeID, masterSrv.Port())
+	clusterNodes = fmt.Sprintf("%s\n%s 127.0.0.1 %d slave %s", clusterNodes, replicaNodeID, replicaSrv.Port(), masterNodeID)
+
+	require.NoError(t, masterClient.Do(ctx, "clusterx", "SETNODES", clusterNodes, "1").Err())
+	require.NoError(t, replicaClient.Do(ctx, "clusterx", "SETNODES", clusterNodes, "1").Err())
+
+	t.Run("Cluster replication should work", func(t *testing.T) {
+		util.WaitForSync(t, replicaClient)
+		require.Equal(t, "slave", util.FindInfoEntry(replicaClient, "role"))
+		masterClient.Set(ctx, "k0", "v0", 0)
+		masterClient.LPush(ctx, "k1", "e0", "e1", "e2")
+		util.WaitForOffsetSync(t, masterClient, replicaClient)
+
+		require.Equal(t, "v0", replicaClient.Get(ctx, "k0").Val())
+		require.Equal(t, []string{"e2", "e1", "e0"}, replicaClient.LRange(ctx, "k1", 0, -1).Val())
+	})
+
+	t.Run("Cluster replication should work normally after restart(issue #2214)", func(t *testing.T) {
+		replicaSrv.Close()
+		masterClient.Set(ctx, "k0", "v1", 0)
+		masterClient.HSet(ctx, "k2", "f0", "v0", "f1", "v1")
+
+		// start the replica server again
+		replicaSrv.Start()
+		_ = replicaClient.Close()
+		replicaClient = replicaSrv.NewClient()
+		// allow to run the read-only command in the replica
+		require.NoError(t, replicaClient.ReadOnly(ctx).Err())
+
+		util.WaitForOffsetSync(t, masterClient, replicaClient)
+		require.Equal(t, "v1", replicaClient.Get(ctx, "k0").Val())
+		require.Equal(t, map[string]string{"f0": "v0", "f1": "v1"}, replicaClient.HGetAll(ctx, "k2").Val())
+	})
+}
+
 func TestReplicationWithHostname(t *testing.T) {
 	srvA := util.StartServer(t, map[string]string{})
 	defer srvA.Close()

tests/gocase/util/server.go

@@ -136,7 +136,10 @@ func (s *KvrocksServer) close(keepDir bool) {
 
 func (s *KvrocksServer) Restart() {
 	s.close(true)
+	s.Start()
+}
 
+func (s *KvrocksServer) Start() {
 	b := *binPath
 	require.NotEmpty(s.t, b, "please set the binary path by `-binPath`")
 	cmd := exec.Command(b)