Moved commit "Ignore message acks from previous subsessions" from ydb repo

qyryq · qyryq · commit 12b4b8e19e87 · 2024-08-26T17:53:29.000Z
diff --git a/src/client/federated_topic/impl/federated_write_session.cpp b/src/client/federated_topic/impl/federated_write_session.cpp
@@ -139,9 +139,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
`@@ -139,9 +139,13 @@ std::shared_ptr<NTopic::IWriteSession> TFederatedWriteSessionImpl::OpenSubsessio`
`139`	`139`	`}`
`140`	`140`	`}`
`141`	`141`	`})`
`142`		`- .AcksHandler([selfCtx = SelfContext](NTopic::TWriteSessionEvent::TAcksEvent& ev) {`
	`142`	`+ .AcksHandler([selfCtx = SelfContext, generation = SubsessionGeneration](NTopic::TWriteSessionEvent::TAcksEvent& ev) {`
`143`	`143`	`if (auto self = selfCtx->LockShared()) {`
`144`	`144`	`with_lock(self->Lock) {`
	`145`	`+ if (generation != self->SubsessionGeneration) {`
	`146`	`+ return;`
	`147`	`+ }`
	`148`	`+`
`145`	`149`	`Y_ABORT_UNLESS(ev.Acks.size() <= self->OriginalMessagesToGetAck.size());`
`146`	`150`
`147`	`151`	`for (size_t i = 0; i < ev.Acks.size(); ++i) {`