feat: populate client_stats in AppendRowsRequest

Author: agrawal-siddharth

java-bigquerystorage/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

@@ -1289,6 +1289,11 @@ private void appendLoop() {
         }
         firstRequestForTableOrSchemaSwitch = false;
 
+        ClientStats clientStats = createClientStats(wrapper);
+        if (!clientStats.equals(ClientStats.getDefaultInstance())) {
+          originalRequestBuilder.setClientStats(clientStats);
+        }
+
         requestProfilerHook.startOperation(
             RequestProfiler.OperationName.RESPONSE_LATENCY, requestUniqueId);
 
@@ -1307,6 +1312,20 @@ private void appendLoop() {
     cleanupConnectionAndRequests(/* avoidBlocking= */ false);
   }
 
+  private ClientStats createClientStats(AppendRequestAndResponse requestWrapper) {
+    ClientStats.Builder clientStatsBuilder = ClientStats.newBuilder();
+    ClientStats.RequestStats.Builder requestStatsBuilder = ClientStats.RequestStats.newBuilder();
+    boolean requestStatsModified = false;
+    if (requestWrapper.requestSendTimeStamp != null) {
+      requestStatsBuilder.setSendTimeMillis(requestWrapper.requestSendTimeStamp.toEpochMilli());
+      requestStatsModified = true;
+    }
+    if (requestStatsModified) {
+      clientStatsBuilder.setRequestStats(requestStatsBuilder.build());
+    }
+    return clientStatsBuilder.build();
+  }
+
   @Nullable
   private AppendRowsSchema getSchema(AppendRowsRequest request) {
     if (request.hasProtoRows()) {

java-bigquerystorage/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/ConnectionWorkerTest.java

@@ -1722,6 +1722,11 @@ void testEarliestSendTime_retryScenario() throws Exception {
 
       // Once exhausted, state should be null
       assertThat(connectionWorker.getEarliestSendTime()).isNull();
+
+      AppendRowsRequest initialRequest = testBigQueryWrite.getAppendRequests().get(0);
+      AppendRowsRequest retryRequest = testBigQueryWrite.getAppendRequests().get(1);
+      assertThat(retryRequest.getClientStats().getRequestStats().getSendTimeMillis())
+          .isAtLeast(initialRequest.getClientStats().getRequestStats().getSendTimeMillis());
     }
   }
 }