Return WriteResponse from WriteAsync (#117)

src/Knet.Kudu.Client/KuduClient.cs

@@ -392,7 +392,7 @@ public async Task<KuduTable> OpenTableAsync(
             return new KuduTable(response);
         }
 
-        public async Task WriteAsync(
+        public async Task<WriteResponse> WriteAsync(
             IEnumerable<KuduOperation> operations,
             ExternalConsistencyMode externalConsistencyMode = ExternalConsistencyMode.ClientPropagated,
             CancellationToken cancellationToken = default)
@@ -428,15 +428,28 @@ public async Task WriteAsync(
             }
 
             var results = await Task.WhenAll(tasks).ConfigureAwait(false);
+
+            return CreateWriteResponse(results);
+        }
+
+        private WriteResponse CreateWriteResponse(WriteResponsePB[] results)
+        {
+            long writeTimestamp = NoTimestamp;
             List<KuduStatus> rowErrors = null;
 
             foreach (var result in results)
             {
+                var timestamp = (long)result.Timestamp;
                 var perRowErrors = result.PerRowErrors;
 
+                if (writeTimestamp == NoTimestamp || writeTimestamp < timestamp)
+                {
+                    writeTimestamp = timestamp;
+                }
+
                 if (perRowErrors.Count > 0)
                 {
-                    rowErrors = rowErrors ?? new List<KuduStatus>();
+                    rowErrors ??= new List<KuduStatus>();
 
                     foreach (var rowError in perRowErrors)
                     {
@@ -450,6 +463,8 @@ public async Task WriteAsync(
             {
                 throw new KuduWriteException(rowErrors);
             }
+
+            return new WriteResponse(writeTimestamp);
         }
 
         private async Task<WriteResponsePB> WriteAsync(

src/Knet.Kudu.Client/WriteResponse.cs

@@ -0,0 +1,15 @@
+namespace Knet.Kudu.Client
+{
+    public class WriteResponse
+    {
+        /// <summary>
+        /// The HybridTime-encoded write timestamp.
+        /// </summary>
+        public long WriteTimestampRaw { get; }
+
+        public WriteResponse(long writeTimestampRaw)
+        {
+            WriteTimestampRaw = writeTimestampRaw;
+        }
+    }
+}

test/Knet.Kudu.Client.FunctionalTests/ClientStressTests.cs

@@ -22,7 +22,6 @@ public class ClientStressTests : IAsyncLifetime
         private KuduClient _client;
         private TestConnectionFactory _testConnectionFactory;
         private KuduTable _table;
-        private long _sharedWriteTimestamp;
 
         public async Task InitializeAsync()
         {
@@ -154,8 +153,6 @@ ValueTask HandleSessionExceptionAsync(SessionExceptionContext context)
                     flush = ThreadSafeRandom.Instance.NextBool();
 
                 currentRowKey++;
-
-                Volatile.Write(ref _sharedWriteTimestamp, _client.LastPropagatedTimestamp);
             }
 
             await session.FlushAsync();
@@ -171,9 +168,9 @@ private async Task ScanAsync(CancellationToken cancellationToken)
 
             while (!cancellationToken.IsCancellationRequested)
             {
-                var sharedWriteTimestamp = Volatile.Read(ref _sharedWriteTimestamp);
+                var timestamp = _client.LastPropagatedTimestamp;
 
-                if (sharedWriteTimestamp == 0)
+                if (timestamp == 0)
                 {
                     // Nothing has been written yet.
                 }
@@ -235,11 +232,11 @@ void AccumulateResults(ResultSet resultSet)
 
         private KuduScannerBuilder GetScannerBuilder()
         {
-            var sharedWriteTimestamp = Volatile.Read(ref _sharedWriteTimestamp);
+            var timestamp = _client.LastPropagatedTimestamp;
 
             return _client.NewScanBuilder(_table)
                 .SetReadMode(ReadMode.ReadAtSnapshot)
-                .SnapshotTimestampRaw(sharedWriteTimestamp)
+                .SnapshotTimestampRaw(timestamp)
                 .SetFaultTolerant(true);
         }