Investigate heartbeat problems

Author: snaury

ydb/core/change_exchange/change_record.h

@@ -31,6 +31,7 @@ class IChangeRecord: public TThrRefBase {
         AsyncIndex,
         CdcDataChange,
         CdcHeartbeat,
+        CdcHeartbeatPrivate,
     };
 
 public:

ydb/core/tx/datashard/cdc_stream_heartbeat.cpp

@@ -45,7 +45,8 @@ class TDataShard::TTxCdcStreamEmitHeartbeats: public NTabletFlatExecutor::TTrans
         const auto heartbeats = Self->GetCdcStreamHeartbeatManager().EmitHeartbeats(txc.DB, Edge);
 
         for (const auto& [streamPathId, info] : heartbeats) {
-            auto recordPtr = TChangeRecordBuilder(TChangeRecord::EKind::CdcHeartbeat)
+        for (auto kind : {TChangeRecord::EKind::CdcHeartbeatPrivate, TChangeRecord::EKind::CdcHeartbeat}) {
+            auto recordPtr = TChangeRecordBuilder(kind)
                 .WithOrder(Self->AllocateChangeRecordOrder(db))
                 .WithGroup(0)
                 .WithStep(info.Last.Step)
@@ -69,6 +70,7 @@ class TDataShard::TTxCdcStreamEmitHeartbeats: public NTabletFlatExecutor::TTrans
                 .SchemaVersion = record.GetSchemaVersion(),
             });
         }
+        }
 
         return true;
     }

ydb/core/tx/datashard/change_record.cpp

@@ -25,6 +25,9 @@ void TChangeRecord::Serialize(NKikimrChangeExchange::TChangeRecord& record) cons
         case EKind::CdcHeartbeat: {
             break;
         }
+        case EKind::CdcHeartbeatPrivate: {
+            break;
+        }
     }
 }
 
@@ -54,6 +57,10 @@ TConstArrayRef<TCell> TChangeRecord::GetKey() const {
         case EKind::CdcHeartbeat: {
             Y_ABORT("Not supported");
         }
+
+        case EKind::CdcHeartbeatPrivate: {
+            Y_ABORT("Not supported");
+        }
     }
 
     Y_ABORT_UNLESS(Key);
@@ -74,6 +81,7 @@ TInstant TChangeRecord::GetApproximateCreationDateTime() const {
 bool TChangeRecord::IsBroadcast() const {
     switch (Kind) {
         case EKind::CdcHeartbeat:
+        case EKind::CdcHeartbeatPrivate:
             return true;
         default:
             return false;

ydb/core/tx/datashard/change_record_cdc_serializer.cpp

@@ -57,6 +57,7 @@ class TBaseSerializer: public IChangeRecordSerializer {
         cmd.SetCreateTimeMS(record.GetApproximateCreationDateTime().MilliSeconds());
         switch (record.GetKind()) {
         case TChangeRecord::EKind::CdcDataChange:
+        case TChangeRecord::EKind::CdcHeartbeatPrivate:
             return SerializeDataChange(cmd, record);
         case TChangeRecord::EKind::CdcHeartbeat:
             return SerializeHeartbeat(cmd, record);
@@ -294,6 +295,12 @@ class TYdbJsonSerializer: public TJsonSerializer {
             return SerializeVirtualTimestamp(json["resolved"], {record.GetStep(), record.GetTxId()});
         }
 
+        json["shard_id"] = Opts.ShardId;
+
+        if (record.GetKind() == TChangeRecord::EKind::CdcHeartbeatPrivate) {
+            return SerializeVirtualTimestamp(json["resolved_private"], {record.GetStep(), record.GetTxId()});
+        }
+
         Y_ABORT_UNLESS(record.GetSchema());
         const auto body = ParseBody(record.GetBody());