Fix retry logic (#182)

Add retry logic to GetMasterLeaderInfoAsync() These methods were affected, - GetHiveMetastoreConfigAsync - GetLocationAsync - GetClusterIdAsync - ExportAuthenticationCredentialsAsync - FindLeaderMasterServerAsync (only used by tests) Fixes a regression introduced by #166 This PR also optimizes SendRpcToMasterAsync() and SendRpcToTxnManagerAsync().
xqrzd · Sep 18, 2021 · 34537c0 · 34537c0
1 parent 4866fd2
commit 34537c0
Showing 1 changed file with 44 additions and 15 deletions.
diff --git a/src/Knet.Kudu.Client/KuduClient.cs b/src/Knet.Kudu.Client/KuduClient.cs
@@ -1223,8 +1223,15 @@ private ValueTask<MasterLeaderInfo> GetMasterLeaderInfoAsync(
             return new ValueTask<MasterLeaderInfo>(masterLeaderInfo);
         }
 
-        var task = ConnectToClusterAsync(cancellationToken);
+        var task = ConnectAsync(cancellationToken);
         return new ValueTask<MasterLeaderInfo>(task);
+
+        async Task<MasterLeaderInfo> ConnectAsync(CancellationToken cancellationToken)
+        {
+            var rpc = new ConnectToMasterRequest();
+            await SendRpcAsync(rpc, cancellationToken).ConfigureAwait(false);
+            return _masterLeaderInfo;
+        }
     }
 
     /// <summary>
@@ -1431,7 +1438,7 @@ internal async Task<T> SendRpcAsync<T>(
                     {
                         KuduMasterRpc<T> masterRpc => SendRpcToMasterAsync(masterRpc, token),
                         KuduTabletRpc<T> tabletRpc => SendRpcToTabletAsync(tabletRpc, token),
-                        KuduTxnRpc<T> txnRpc => SendRpcToTxnAsync(txnRpc, token),
+                        KuduTxnRpc<T> txnRpc => SendRpcToTxnManagerAsync(txnRpc, token),
                         _ => throw new NotSupportedException()
                     };
 
@@ -1485,28 +1492,50 @@ private void CompleteTrackedRpc(KuduRpc rpc)
         }
     }
 
-    private async Task<T> SendRpcToMasterAsync<T>(
+    private Task<T> SendRpcToMasterAsync<T>(
         KuduMasterRpc<T> rpc, CancellationToken cancellationToken)
     {
-        var masterLeaderInfo = await GetMasterLeaderInfoAsync(cancellationToken)
-            .ConfigureAwait(false);
+        var masterLeaderInfo = _masterLeaderInfo;
+        if (masterLeaderInfo is not null)
+        {
+            var serverInfo = masterLeaderInfo.ServerInfo;
+            return SendRpcToMasterAsync(rpc, serverInfo, cancellationToken);
+        }
 
-        var serverInfo = masterLeaderInfo.ServerInfo;
+        return SendAsync(rpc, cancellationToken);
 
-        return await SendRpcToMasterAsync(rpc, serverInfo, cancellationToken)
-            .ConfigureAwait(false);
+        async Task<T> SendAsync(KuduMasterRpc<T> rpc, CancellationToken cancellationToken)
+        {
+            var masterLeaderInfo = await ConnectToClusterAsync(cancellationToken)
+                .ConfigureAwait(false);
+
+            var serverInfo = masterLeaderInfo.ServerInfo;
+            return await SendRpcToMasterAsync(rpc, serverInfo, cancellationToken)
+                .ConfigureAwait(false);
+        }
     }
 
-    private async Task<T> SendRpcToTxnAsync<T>(
+    private Task<T> SendRpcToTxnManagerAsync<T>(
         KuduTxnRpc<T> rpc, CancellationToken cancellationToken)
     {
-        var masterLeaderInfo = await GetMasterLeaderInfoAsync(cancellationToken)
-            .ConfigureAwait(false);
+        var masterLeaderInfo = _masterLeaderInfo;
+        if (masterLeaderInfo is not null)
+        {
+            var serverInfo = masterLeaderInfo.ServerInfo;
+            return SendRpcToTxnManagerAsync(rpc, serverInfo, cancellationToken);
+        }
 
-        var serverInfo = masterLeaderInfo.ServerInfo;
+        return SendAsync(rpc, cancellationToken);
 
-        return await SendRpcToTxnAsync(rpc, serverInfo, cancellationToken)
-            .ConfigureAwait(false);
+        async Task<T> SendAsync(KuduTxnRpc<T> rpc, CancellationToken cancellationToken)
+        {
+            var masterLeaderInfo = await ConnectToClusterAsync(cancellationToken)
+                .ConfigureAwait(false);
+
+            var serverInfo = masterLeaderInfo.ServerInfo;
+            return await SendRpcToTxnManagerAsync(rpc, serverInfo, cancellationToken)
+                .ConfigureAwait(false);
+        }
     }
 
     /// <summary>
@@ -1617,7 +1646,7 @@ await SendRpcToServerGenericAsync(rpc, serverInfo, cancellationToken)
         return rpc.Output;
     }
 
-    private async Task<T> SendRpcToTxnAsync<T>(
+    private async Task<T> SendRpcToTxnManagerAsync<T>(
         KuduTxnRpc<T> rpc,
         ServerInfo serverInfo,
         CancellationToken cancellationToken)