generic lookup: enable tcp keep-alive on connector connection (backport #13118)

ydb/library/grpc/client/grpc_client_low.cpp

@@ -141,15 +141,8 @@ void TChannelPool::GetStubsHolderLocked(
                 }
             }
         }
-        TGRpcKeepAliveSocketMutator* mutator = nullptr;
+        auto mutator = NImpl::CreateGRpcKeepAliveSocketMutator(TcpKeepAliveSettings_);
         // will be destroyed inside grpc
-        if (TcpKeepAliveSettings_.Enabled) {
-            mutator = new TGRpcKeepAliveSocketMutator(
-                TcpKeepAliveSettings_.Idle,
-                TcpKeepAliveSettings_.Count,
-                TcpKeepAliveSettings_.Interval
-            );
-        }
         cb(Pool_.emplace(channelId, CreateChannelInterface(config, mutator)).first->second);
         LastUsedQueue_.emplace(Pool_.at(channelId).GetLastUseTime(), channelId);
     }
@@ -588,4 +581,16 @@ void TGRpcClientLow::ForgetContext(TContextImpl* context) {
     }
 }
 
+grpc_socket_mutator* NImpl::CreateGRpcKeepAliveSocketMutator(const TTcpKeepAliveSettings& TcpKeepAliveSettings_) {
+    TGRpcKeepAliveSocketMutator* mutator = nullptr;
+    if (TcpKeepAliveSettings_.Enabled) {
+        mutator = new TGRpcKeepAliveSocketMutator(
+                TcpKeepAliveSettings_.Idle,
+                TcpKeepAliveSettings_.Count,
+                TcpKeepAliveSettings_.Interval
+                );
+    }
+    return mutator;
+}
+
 } // namespace NGRpc

ydb/library/grpc/client/grpc_client_low.h

@@ -451,6 +451,10 @@ class IStreamRequestReadWriteProcessor : public IStreamRequestReadProcessor<TRes
 
 class TGRpcKeepAliveSocketMutator;
 
+namespace NImpl {
+grpc_socket_mutator* CreateGRpcKeepAliveSocketMutator(const TTcpKeepAliveSettings& TcpKeepAliveSettings_);
+} // NImpl
+
 // Class to hold stubs allocated on channel.
 // It is poor documented part of grpc. See KIKIMR-6109 and comment to this commit
 
@@ -1386,6 +1390,13 @@ class TGRpcClientLow
         return std::unique_ptr<TServiceConnection<TGRpcService>>(new TServiceConnection<TGRpcService>(CreateChannelInterface(config), this));
     }
 
+    template<typename TGRpcService>
+    std::unique_ptr<TServiceConnection<TGRpcService>> CreateGRpcServiceConnection(const TGRpcClientConfig& config, const TTcpKeepAliveSettings& keepAlive) {
+        auto mutator = NImpl::CreateGRpcKeepAliveSocketMutator(keepAlive);
+        // will be destroyed inside grpc
+        return std::unique_ptr<TServiceConnection<TGRpcService>>(new TServiceConnection<TGRpcService>(CreateChannelInterface(config, mutator), this));
+    }
+
     template<typename TGRpcService>
     std::unique_ptr<TServiceConnection<TGRpcService>> CreateGRpcServiceConnection(TStubsHolder& holder) {
         return std::unique_ptr<TServiceConnection<TGRpcService>>(new TServiceConnection<TGRpcService>(holder, this));

ydb/library/yql/providers/generic/connector/libcpp/client.cpp

@@ -44,7 +44,13 @@ namespace NYql::NConnector {
             GrpcClient_ = std::make_unique<NYdbGrpc::TGRpcClientLow>();
 
             // FIXME: is it OK to use single connection during the client lifetime?
-            GrpcConnection_ = GrpcClient_->CreateGRpcServiceConnection<NApi::Connector>(GrpcConfig_);
+            GrpcConnection_ = GrpcClient_->CreateGRpcServiceConnection<NApi::Connector>(GrpcConfig_, NYdbGrpc::TTcpKeepAliveSettings {
+                    // TODO configure hardcoded values
+                    .Enabled = true,
+                    .Idle = 30,
+                    .Count = 5,
+                    .Interval = 10
+            });
         }
 
         virtual TDescribeTableAsyncResult DescribeTable(const NApi::TDescribeTableRequest& request) override {