fix: revise health metrics inflightBytes to become inflightBytesMax (#14613)

Author: agrawal-siddharth

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

@@ -326,6 +326,9 @@ class HealthCheckMetrics {
     @GuardedBy("lock")
     private long windowedQueuedRetriesMax;
 
+    @GuardedBy("lock")
+    private long windowedInflightBytesMax;
+
     private long windowedConnectionAttemptCount;
     private long windowedConnectionClosedCount;
 
@@ -338,6 +341,13 @@ void updateWindowedQueuedRequestsMax(long currentQueueLength, long currentRetryC
       }
     }
 
+    @GuardedBy("lock")
+    void updateInflightBytesMax(long currentInflightBytes) {
+      if (currentInflightBytes > windowedInflightBytesMax) {
+        windowedInflightBytesMax = currentInflightBytes;
+      }
+    }
+
     void updateResponseWait(Instant sendInstant) {
       long currentWaitTime = Duration.between(sendInstant, Instant.now()).toMillis();
       if (currentWaitTime > windowedMilliResponseWaitTimeMax) {
@@ -380,7 +390,7 @@ class HealthCheckFields {
       long responseCount;
       long queuedRequestCountMax;
       long queuedRetryCountMax; // How many active waiting or inflight requests are retries
-      long inflightBytes;
+      long inflightBytesMax;
       long connectionAttemptCount;
       long connectionClosedCount;
       boolean isConnected; // snapshot at instant metrics are gathered
@@ -399,7 +409,7 @@ private void gatherHealthCheckMetrics(HealthCheckFields healthCheckFields) {
       healthCheckFields.queuedRequestCountMax = windowedQueuedRequestsMax;
       healthCheckFields.queuedRetryCountMax = windowedQueuedRetriesMax;
       healthCheckFields.msecLongestResponseWaitTime = windowedMilliResponseWaitTimeMax;
-      healthCheckFields.inflightBytes = inflightBytes.get();
+      healthCheckFields.inflightBytesMax = windowedInflightBytesMax;
       healthCheckFields.requestsSentCount = windowedRequestsSent;
       healthCheckFields.responseCount = windowedResponsesAcked;
       if (HEALTH_CHECK_INTERVAL.toMillis() > 0) {
@@ -425,7 +435,7 @@ private void gatherHealthCheckMetrics(HealthCheckFields healthCheckFields) {
      */
     private boolean checkThresholds(HealthCheckFields healthCheckFields) {
       if ((healthCheckFields.queuedRequestCountMax >= queuedRequestsThreshold)
-          || (healthCheckFields.inflightBytes >= queuedBytesThreshold)
+          || (healthCheckFields.inflightBytesMax >= queuedBytesThreshold)
           || (healthCheckFields.msecLongestResponseWaitTime >= responseWaitTimeThreshold.toMillis())
           || (healthCheckFields.msecMaxLatency >= latencyThreshold.toMillis())
           || (healthCheckFields.connectionAttemptCount >= connectionAttemptThreshold)
@@ -479,6 +489,7 @@ private void resetWindowedMetrics() {
       windowedConnectionClosedCount = 0;
       windowedQueuedRequestsMax = 0;
       windowedQueuedRetriesMax = 0;
+      windowedInflightBytesMax = 0;
     }
 
     /*
@@ -785,6 +796,7 @@ private void addMessageToWaitingQueue(
       AppendRequestAndResponse requestWrapper, boolean addToFront) {
     this.inflightRequests.incrementAndGet();
     this.inflightBytes.addAndGet(requestWrapper.messageSize);
+    healthCheckMetrics.updateInflightBytesMax(this.inflightBytes.get());
     hasMessageInWaitingQueue.signal();
     requestProfilerHook.startOperation(
         RequestProfiler.OperationName.WAIT_QUEUE, requestWrapper.requestUniqueId);
@@ -932,6 +944,7 @@ private ApiFuture<AppendRowsResponse> appendInternal(
       requestProfilerHook.startOperation(RequestProfiler.OperationName.WAIT_QUEUE, requestUniqueId);
       this.inflightRequests.incrementAndGet();
       this.inflightBytes.addAndGet(requestWrapper.messageSize);
+      healthCheckMetrics.updateInflightBytesMax(this.inflightBytes.get());
       requestWrapper.placedInWaitingQueueTime = Instant.now();
       waitingRequestQueue.addLast(requestWrapper);
       healthCheckMetrics.updateWindowedQueuedRequestsMax(

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

@@ -1296,7 +1296,7 @@ void testHealthCheck() throws Exception {
         healthCheckFields.msecLongestResponseWaitTime > 1
             && healthCheckFields.msecLongestResponseWaitTime < msecResponseDelay);
     assertEquals(appendCount, healthCheckFields.queuedRequestCountMax);
-    assertEquals(appendCount * sizePerRequest, healthCheckFields.inflightBytes);
+    assertEquals(appendCount * sizePerRequest, healthCheckFields.inflightBytesMax);
     assertEquals("MULTIPLEXING", healthCheckFields.streamName);
     assertEquals(connectionWorker.getWriterId(), healthCheckFields.writerId);
 
@@ -1363,7 +1363,7 @@ void testHealthCheckThresholds() throws Exception {
     fields.responseCount = 0;
     fields.queuedRequestCountMax = 0;
     fields.queuedRetryCountMax = 0;
-    fields.inflightBytes = 0;
+    fields.inflightBytesMax = 0;
     fields.connectionAttemptCount = 0;
     fields.connectionClosedCount = 0;
     fields.isConnected = false;
@@ -1415,9 +1415,9 @@ void testHealthCheckThresholds() throws Exception {
     fields.queuedRetryCountMax = 0;
 
     // inflightBytes >= queuedBytesThreshold (52428800)
-    fields.inflightBytes = 50 * 1024 * 1024;
+    fields.inflightBytesMax = 50 * 1024 * 1024;
     assertEquals(true, connectionWorker.checkTestOnlyHealthCheckThresholds(fields));
-    fields.inflightBytes = 0;
+    fields.inflightBytesMax = 0;
 
     // connectionAttemptCount >= connectionAttemptThreshold (1)
     fields.connectionAttemptCount = 1;