[HUDI-8701] prevent lock metrics failure from failing process

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/metrics/HoodieLockMetrics.java

@@ -19,6 +19,7 @@
 package org.apache.hudi.client.transaction.lock.metrics;
 
 import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.metrics.Metrics;
 import org.apache.hudi.storage.HoodieStorage;
@@ -27,10 +28,13 @@
 import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.SlidingWindowReservoir;
 import com.codahale.metrics.Timer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.TimeUnit;
 
 public class HoodieLockMetrics {
+  private static final Logger LOG = LoggerFactory.getLogger(HoodieLockMetrics.class);
 
   public static final String LOCK_ACQUIRE_ATTEMPTS_COUNTER_NAME = "lock.acquire.attempts";
   public static final String LOCK_ACQUIRE_SUCCESS_COUNTER_NAME = "lock.acquire.success";
@@ -93,8 +97,12 @@ public void startLockApiTimerContext() {
 
   public void updateLockAcquiredMetric() {
     if (isMetricsEnabled) {
-      long durationMs = lockApiRequestDurationTimer.endTimer();
-      lockApiRequestDuration.update(durationMs, TimeUnit.MILLISECONDS);
+      Option<Long> durationMs = lockApiRequestDurationTimer.tryEndTimer();
+      if (durationMs.isPresent()) {
+        lockApiRequestDuration.update(durationMs.get(), TimeUnit.MILLISECONDS);
+      } else {
+        LOG.info("Unable to get lock request duration");
+      }
       lockAttempts.inc();
       successfulLockAttempts.inc();
       lockDurationTimer.startTimer();
@@ -103,16 +111,24 @@ public void updateLockAcquiredMetric() {
 
   public void updateLockNotAcquiredMetric() {
     if (isMetricsEnabled) {
-      long durationMs = lockApiRequestDurationTimer.endTimer();
-      lockApiRequestDuration.update(durationMs, TimeUnit.MILLISECONDS);
+      Option<Long> durationMs = lockApiRequestDurationTimer.tryEndTimer();
+      if (durationMs.isPresent()) {
+        lockApiRequestDuration.update(durationMs.get(), TimeUnit.MILLISECONDS);
+      } else {
+        LOG.info("Unable to get lock request duration");
+      }
       failedLockAttempts.inc();
     }
   }
 
   public void updateLockHeldTimerMetrics() {
     if (isMetricsEnabled && lockDurationTimer != null) {
-      long lockDurationInMs = lockDurationTimer.endTimer();
-      lockDuration.update(lockDurationInMs, TimeUnit.MILLISECONDS);
+      Option<Long> lockDurationInMs = lockDurationTimer.tryEndTimer();
+      if (lockDurationInMs.isPresent()) {
+        lockDuration.update(lockDurationInMs.get(), TimeUnit.MILLISECONDS);
+      } else {
+        LOG.info("Unable to get lock duration");
+      }
     }
   }
 }

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestHoodieLockMetrics.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.client.transaction;
+
+import org.apache.hudi.client.transaction.lock.metrics.HoodieLockMetrics;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.config.metrics.HoodieMetricsConfig;
+import org.apache.hudi.metrics.MetricsReporterType;
+import org.apache.hudi.storage.HoodieStorage;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.mockito.Mockito.mock;
+
+public class TestHoodieLockMetrics {
+
+  @Test
+  public void testMetricsHappyPath() {
+    HoodieStorage storage = mock(HoodieStorage.class);
+    HoodieMetricsConfig metricsConfig = HoodieMetricsConfig.newBuilder().withPath("/gdsafsd")
+        .withReporterType(MetricsReporterType.INMEMORY.name()).withLockingMetrics(true).build();
+    HoodieLockMetrics lockMetrics = new HoodieLockMetrics(HoodieWriteConfig.newBuilder()
+        .forTable("idk").withPath("/dsfasdf/asdf")
+        .withMetricsConfig(metricsConfig)
+        .build(), storage);
+
+    //lock acquired
+    assertDoesNotThrow(lockMetrics::startLockApiTimerContext);
+    assertDoesNotThrow(lockMetrics::updateLockAcquiredMetric);
+    assertDoesNotThrow(lockMetrics::updateLockHeldTimerMetrics);
+
+    //lock not acquired
+    assertDoesNotThrow(lockMetrics::startLockApiTimerContext);
+    assertDoesNotThrow(lockMetrics::updateLockNotAcquiredMetric);
+  }
+
+  @Test
+  public void testMetricsMisses() {
+    HoodieStorage storage = mock(HoodieStorage.class);
+    HoodieMetricsConfig metricsConfig = HoodieMetricsConfig.newBuilder().withPath("/gdsafsd")
+        .withReporterType(MetricsReporterType.INMEMORY.name()).withLockingMetrics(true).build();
+    HoodieLockMetrics lockMetrics = new HoodieLockMetrics(HoodieWriteConfig.newBuilder()
+        .forTable("idk").withPath("/dsfasdf/asdf")
+        .withMetricsConfig(metricsConfig)
+        .build(), storage);
+
+    assertDoesNotThrow(lockMetrics::updateLockHeldTimerMetrics);
+    assertDoesNotThrow(lockMetrics::updateLockNotAcquiredMetric);
+    assertDoesNotThrow(lockMetrics::updateLockAcquiredMetric);
+  }
+
+}

hudi-common/src/test/java/org/apache/hudi/common/util/TestHoodieTimer.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.common.util;
+
+import org.junit.jupiter.api.Test;
+
+import static java.lang.Thread.sleep;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestHoodieTimer {
+
+  @Test
+  public void testTryStop() throws InterruptedException {
+    HoodieTimer timer = new HoodieTimer();
+    assertFalse(timer.tryEndTimer().isPresent());
+
+    timer.startTimer();
+    sleep(250);
+    Option<Long> result = timer.tryEndTimer();
+    assertTrue(result.isPresent());
+    assertTrue(result.get() >= 250);
+  }
+}

hudi-io/src/main/java/org/apache/hudi/common/util/HoodieTimer.java

@@ -83,6 +83,17 @@ public long endTimer() {
     return timeInfoDeque.pop().stop();
   }
 
+  /**
+   * tries to end the timer but will not cause a runtime failure like endTimer will if the queue is empty
+   * @return option of the timer duration, empty option if the queue is empty
+   */
+  public Option<Long> tryEndTimer() {
+    if (timeInfoDeque.isEmpty()) {
+      return Option.empty();
+    }
+    return Option.of(timeInfoDeque.pop().stop());
+  }
+
   /**
    * Creates an instance of {@link HoodieTimer} already started
    */