ydb_topic: Call non-dataEvent callbacks immediately (#10751)

Author: qyryq

ydb/public/sdk/cpp/client/ydb_topic/impl/read_session_impl.ipp

@@ -1955,7 +1955,7 @@ void TSingleClusterReadSessionImpl<UseMigrationProtocol>::TrySubscribeOnTransact
 
         auto callback = [cbContext = this->SelfContext, txId, txInfo, consumer = Settings.ConsumerName_, client]() {
             TVector<TTopicOffsets> offsets;
-            
+
             with_lock (txInfo->Lock) {
                 Y_ABORT_UNLESS(!txInfo->CommitCalled);
 
@@ -2090,6 +2090,13 @@ bool TReadSessionEventsQueue<UseMigrationProtocol>::PushEvent(TIntrusivePtr<TPar
             stream->DeleteNotReadyTail(deferred);
         }
 
+        if (!HasDataEventCallback() && !std::holds_alternative<TADataReceivedEvent<UseMigrationProtocol>>(event)) {
+            // Call non-dataEvent callbacks immediately.
+            if (TryApplyCallbackToEventImpl(event, deferred, CbContext)) {
+                return true;
+            }
+        }
+
         stream->InsertEvent(std::move(event));
         Y_ASSERT(stream->HasEvents());