Ignore message acks from previous subsessions (ydb-platform#7683)

qyryq · web-flow · commit 0b61b05de7ce · 2024-08-13T17:00:03.000+03:00
diff --git a/ydb/public/sdk/cpp/client/ydb_federated_topic/impl/federated_write_session.cpp b/ydb/public/sdk/cpp/client/ydb_federated_topic/impl/federated_write_session.cpp
@@ -140,9 +140,13 @@ std::shared_ptr<NTopic::IWriteSession> TFederatedWriteSessionImpl::OpenSubsessio
                 }
             }
         })
-        .AcksHandler([selfCtx = SelfContext](NTopic::TWriteSessionEvent::TAcksEvent& ev) {
+        .AcksHandler([selfCtx = SelfContext, generation = SubsessionGeneration](NTopic::TWriteSessionEvent::TAcksEvent& ev) {
             if (auto self = selfCtx->LockShared()) {
                 with_lock(self->Lock) {
+                    if (generation != self->SubsessionGeneration) {
+                        return;
+                    }
+
                     Y_ABORT_UNLESS(ev.Acks.size() <= self->OriginalMessagesToGetAck.size());
 
                     for (size_t i = 0; i < ev.Acks.size(); ++i) {

Original file line number	Diff line number	Diff line change
`@@ -140,9 +140,13 @@ std::shared_ptr<NTopic::IWriteSession> TFederatedWriteSessionImpl::OpenSubsessio`
`140`	`140`	`}`
`141`	`141`	`}`
`142`	`142`	`})`
`143`		`- .AcksHandler([selfCtx = SelfContext](NTopic::TWriteSessionEvent::TAcksEvent& ev) {`
	`143`	`+ .AcksHandler([selfCtx = SelfContext, generation = SubsessionGeneration](NTopic::TWriteSessionEvent::TAcksEvent& ev) {`
`144`	`144`	`if (auto self = selfCtx->LockShared()) {`
`145`	`145`	`with_lock(self->Lock) {`
	`146`	`+ if (generation != self->SubsessionGeneration) {`
	`147`	`+ return;`
	`148`	`+ }`
	`149`	`+`
`146`	`150`	`Y_ABORT_UNLESS(ev.Acks.size() <= self->OriginalMessagesToGetAck.size());`
`147`	`151`
`148`	`152`	`for (size_t i = 0; i < ev.Acks.size(); ++i) {`