Fix ApplyReplicationChanges (#5683)

Author: CyberROFL

ydb/core/tx/datashard/datashard_repl_apply.cpp

@@ -24,13 +24,20 @@ class TDataShard::TTxApplyReplicationChanges : public TTransactionBase<TDataShar
     bool Execute(TTransactionContext& txc, const TActorContext& ctx) override {
         Y_UNUSED(ctx);
 
-        if (Self->State != TShardState::Ready && !Self->IsReplicated()) {
+        if (Self->State != TShardState::Ready) {
             Result = MakeHolder<TEvDataShard::TEvApplyReplicationChangesResult>(
                 NKikimrTxDataShard::TEvApplyReplicationChangesResult::STATUS_REJECTED,
                 NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_WRONG_STATE);
             return true;
         }
 
+        if (!Self->IsReplicated()) {
+            Result = MakeHolder<TEvDataShard::TEvApplyReplicationChangesResult>(
+                NKikimrTxDataShard::TEvApplyReplicationChangesResult::STATUS_REJECTED,
+                NKikimrTxDataShard::TEvApplyReplicationChangesResult::REASON_BAD_REQUEST);
+            return true;
+        }
+
         const auto& msg = Ev->Get()->Record;
 
         const auto& tableId = msg.GetTableId();

ydb/core/tx/datashard/datashard_ut_replication.cpp

@@ -281,6 +281,29 @@ Y_UNIT_TEST_SUITE(DataShardReplication) {
         );
     }
 
+    Y_UNIT_TEST(ApplyChangesToCommonTable) {
+        TPortManager pm;
+        TServerSettings serverSettings(pm.GetPort(2134));
+        serverSettings.SetDomainName("Root")
+            .SetUseRealThreads(false);
+
+        Tests::TServer::TPtr server = new TServer(serverSettings);
+        auto &runtime = *server->GetRuntime();
+        auto sender = runtime.AllocateEdgeActor();
+
+        runtime.SetLogPriority(NKikimrServices::TX_DATASHARD, NLog::PRI_TRACE);
+
+        InitRoot(server, sender);
+        CreateShardedTable(server, sender, "/Root", "table-1", TShardedTableOptions());
+
+        auto shards = GetTableShards(server, sender, "/Root/table-1");
+        auto tableId = ResolveTableId(server, sender, "/Root/table-1");
+
+        ApplyChanges(server, shards.at(0), tableId, "my-source", {
+            TChange{ .Offset = 0, .WriteTxId = 0, .Key = 1, .Value = 11 },
+        }, NKikimrTxDataShard::TEvApplyReplicationChangesResult::STATUS_REJECTED);
+    }
+
 }
 
 } // namespace NKikimr