Moved "Cancel existing Processor in OnConnect" commit from ydb repo

Author: qyryq

src/client/persqueue_public/impl/write_session_impl.cpp

@@ -454,6 +454,10 @@ void TWriteSessionImpl::DoConnect(const TDuration& delay, const std::string& end
         Y_ASSERT(ConnectContext);
         Y_ASSERT(ConnectTimeoutContext);
 
+        if (Processor) {
+            Processor->Cancel();
+        }
+
         // Cancel previous operations.
         Cancel(prevConnectContext);
         if (prevConnectDelayContext)

src/client/topic/impl/read_session_impl.ipp

@@ -279,6 +279,9 @@ bool TSingleClusterReadSessionImpl<UseMigrationProtocol>::Reconnect(const TPlain
             Cancel(connectTimeoutContext);
             return false;
         }
+        if (Processor) {
+            Processor->Cancel();
+        }
         Processor = nullptr;
         WaitingReadResponse = false;
         ServerMessage = std::make_shared<TServerMessage<UseMigrationProtocol>>();

src/client/topic/impl/write_session_impl.cpp

@@ -618,6 +618,10 @@ void TWriteSessionImpl::Connect(const TDuration& delay) {
         }
         Cancel(prevConnectTimeoutContext);
 
+        if (Processor) {
+            Processor->Cancel();
+        }
+
         reqSettings = TRpcRequestSettings::Make(Settings, PreferredPartitionLocation.Endpoint);
 
         connectCallback = [cbContext = SelfContext,