Fix tests

src/Client/ConnectionEstablisher.cpp

@@ -152,6 +152,12 @@ bool ConnectionEstablisherAsync::checkBeforeTaskResume()
     return checkTimeout();
 }
 
+void ConnectionEstablisherAsync::cancelAfter()
+{
+    if (!is_finished)
+        reset();
+}
+
 bool ConnectionEstablisherAsync::checkTimeout()
 {
     bool is_socket_ready = false;

src/Client/ConnectionEstablisher.h

@@ -94,6 +94,8 @@ private:
         void run(AsyncCallback async_callback, ResumeCallback suspend_callback) override;
     };
 
+    void cancelAfter() override;
+
     /// When epoll file descriptor is ready, check if it's an expired timeout.
     /// Return false if receive timeout expired and socket is not ready, return true otherwise.
     bool checkTimeout();

src/Common/AsyncTaskExecutor.cpp

@@ -38,8 +38,9 @@ void AsyncTaskExecutor::cancel()
 {
     std::lock_guard guard(fiber_lock);
     is_cancelled = true;
-    cancelImpl();
+    cancelBefore();
     destroyFiber();
+    cancelAfter();
 }
 
 void AsyncTaskExecutor::restart()

src/Common/AsyncTaskExecutor.h

@@ -93,10 +93,12 @@ protected:
     virtual void processException(std::exception_ptr e) { std::rethrow_exception(e); }
 
     /// Method that is called in cancel() before fiber destruction.
-    virtual void cancelImpl() { }
+    virtual void cancelBefore() { }
+    /// Method that is called in cancel() after fiber destruction.
+    virtual void cancelAfter() { }
 
     /// Resume fiber explicitly without mutex locking.
-    /// Can be called in cancelImpl().
+    /// Can be called in cancelBefore().
     void resumeUnlocked();
 
 private:

src/Processors/examples/protobuf.cpp

@@ -0,0 +1,26 @@
+#include <google/protobuf/compiler/importer.h>
+
+struct ErrorCollector : public google::protobuf::compiler::MultiFileErrorCollector
+{
+    void AddError(const std::string&, int, int, const std::string& msg) override
+    {
+        throw std::runtime_error(msg);
+    }
+};
+
+int main()
+{
+    google::protobuf::compiler::DiskSourceTree disk_source_tree;
+    disk_source_tree.MapPath("", "schemas/");
+    ErrorCollector collector;
+    google::protobuf::compiler::Importer importer(&disk_source_tree, &collector);
+    try
+    {
+        importer.Import("schema.proto");
+    }
+    catch()
+    {
+    }
+
+    importer.pool()->FindMessageTypeByName("Schema");
+}

src/QueryPipeline/RemoteQueryExecutorReadContext.cpp

@@ -104,7 +104,7 @@ bool RemoteQueryExecutorReadContext::checkTimeout(bool blocking)
     return true;
 }
 
-void RemoteQueryExecutorReadContext::cancelImpl()
+void RemoteQueryExecutorReadContext::cancelBefore()
 {
     /// One should not try to wait for the current packet here in case of
     /// timeout because this will exceed the timeout.

src/QueryPipeline/RemoteQueryExecutorReadContext.h

@@ -50,7 +50,7 @@ private:
     void processAsyncEvent(int fd, Poco::Timespan socket_timeout, AsyncEventTimeoutType type, const std::string & description, uint32_t events) override;
     void clearAsyncEvent() override;
 
-    void cancelImpl() override;
+    void cancelBefore() override;
 
     struct Task : public AsyncTask
     {