pq sdk deadlock has been fixed (between Close and OnReadDone) (#3115)

Author: dorooleg

ydb/public/sdk/cpp/client/ydb_persqueue_core/impl/read_session.h

@@ -729,6 +729,10 @@ class TPartitionStreamImpl : public TAPartitionStream<UseMigrationProtocol> {
         EventsQueue.clear();
     }
 
+    TRawPartitionStreamEventQueue<UseMigrationProtocol> ExtractQueue() noexcept {
+        return std::move(EventsQueue);
+    }
+
     static void GetDataEventImpl(TIntrusivePtr<TPartitionStreamImpl<UseMigrationProtocol>> partitionStream,
                                  size_t& maxEventsCount,
                                  size_t& maxByteSize,
@@ -786,14 +790,16 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
 
     bool Close(const TASessionClosedEvent<UseMigrationProtocol>& event, TDeferredActions<UseMigrationProtocol>& deferred) {
         TWaiter waiter;
+        TVector<TRawPartitionStreamEventQueue<UseMigrationProtocol>> defferedDelete;
         with_lock (TParent::Mutex) {
             if (TParent::Closed) {
                 return false;
             }
+            defferedDelete.reserve(TParent::Events.size());
             while (!TParent::Events.empty()) {
                 auto& event = TParent::Events.front();
                 if (!event.IsEmpty()) {
-                    event.PartitionStream->ClearQueue();
+                    defferedDelete.push_back(event.PartitionStream->ExtractQueue());
                 }
                 TParent::Events.pop();
             }
@@ -802,6 +808,9 @@ class TReadSessionEventsQueue: public TBaseSessionEventsQueue<TAReadSessionSetti
             waiter = TWaiter(TParent::Waiter.ExtractPromise(), this);
         }
 
+        // Delayed deletion is necessary to avoid deadlock with PushEvent
+        defferedDelete.clear();
+
         TReadSessionEventInfo<UseMigrationProtocol> info(event);
         ApplyHandler(info, deferred);
         deferred.DeferSignalWaiter(std::move(waiter));