use error reason instead of error message and fix caches

Author: utkarsh-goel-google

lib/build.gradle

@@ -23,7 +23,7 @@ plugins {
 
 // v"2.2.5" is recommended, but it uses Java21 which is incompatible with other
 // plugins (e.g. "com.github.johnrengelman.shadow"), therefore, using latest compatible version
-   id("com.google.cloud.artifactregistry.gradle-plugin") version "2.1.5"
+    id("com.google.cloud.artifactregistry.gradle-plugin") version "2.1.5"
 }
 
 googleJavaFormat {
@@ -62,10 +62,10 @@ dependencies {
     implementation "com.google.api:gax-httpjson:${gaxHttpJsonVersion}"
     implementation "io.grpc:grpc-protobuf:${grpcProtobufVersion}"
     implementation "com.google.guava:guava:${guavaVersion}"
-    
+
     // SLF4J logging
     implementation "org.slf4j:slf4j-api:${slf4jVersion}"
-    
+
     // Lombok for @Slf4j annotation
     compileOnly "org.projectlombok:lombok:${lombokVersion}"
     annotationProcessor "org.projectlombok:lombok:${lombokVersion}"
@@ -74,21 +74,17 @@ dependencies {
     compileOnly "com.google.auto.value:auto-value-annotations:1.10.1"
     annotationProcessor "com.google.auto.value:auto-value:1.10.1"
 
-    // AutoValue
-    compileOnly "com.google.auto.value:auto-value-annotations:1.10.1"
-    annotationProcessor "com.google.auto.value:auto-value:1.10.1"
-    
     // Use JUnit test framework.
     testImplementation "junit:junit:${junitVersion}"
     testImplementation "com.google.truth:truth:${truthVersion}"
     testImplementation "org.mockito:mockito-core:${mockitoVersion}"
     testImplementation "com.google.api:gax:${gaxTestLibVersion}:testlib"
     testImplementation "com.google.testparameterinjector:test-parameter-injector:${parameterInjectorVersion}"
-    
+
     // Test logging dependencies
     testImplementation "ch.qos.logback:logback-classic:${logbackVersion}"
     testImplementation "ch.qos.logback:logback-core:${logbackVersion}"
-    
+
     // Test Lombok
     testCompileOnly "org.projectlombok:lombok:${lombokVersion}"
     testAnnotationProcessor "org.projectlombok:lombok:${lombokVersion}"

lib/src/main/java/com/google/cloud/datalineage/producerclient/CacheOptions.java

@@ -2,7 +2,7 @@
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
-// You may not obtain a copy of the License at
+// You may obtain a copy of the License at
 //
 // https://www.apache.org/licenses/LICENSE-2.0
 //
@@ -18,13 +18,14 @@
 import java.time.Duration;
 
 /**
- * Provides an immutable object for Cache initialization. CacheOptions object
- * can be created via Builder.
+ * Provides an immutable object for Cache initialization. CacheOptions object can be created via
+ * Builder.
  */
 public class CacheOptions {
-  protected static final Duration DEFAULT_DISABLED_TIME = Duration.ofMinutes(10);
+
+  protected static final Duration DEFAULT_DISABLED_TIME = Duration.ofMinutes(5);
   protected static final int DEFAULT_SIZE = 1000;
-  protected static final Clock DEFAULT_CLOCK = Clock.systemUTC();
+  protected static final Clock DEFAULT_CLOCK = Clock.systemDefaultZone();
 
   private final Duration defaultCacheDisabledStatusTime;
   private final int cacheSize;
@@ -60,13 +61,23 @@ public CacheOptions.Builder toBuilder() {
     return new CacheOptions.Builder(this);
   }
 
+
+  @Override
+  public boolean equals(Object options) {
+    return options instanceof CacheOptions that
+        && that.getDefaultCacheDisabledStatusTime() == this.defaultCacheDisabledStatusTime &&
+        that.getCacheSize() == this.cacheSize &&
+        that.clock == this.clock;
+  }
+
   /**
    * * Builder for CacheSettings.
    *
    * <p>Lets setting `markServiceAsDisabledTime`, `cacheSize`, and `clock`. Can be created by
    * CacheOptions.newBuilder method. To create settings object, use build method.
    */
   public static class Builder {
+
     protected Duration defaultCacheDisabledStatusTime;
     protected int cacheSize;
     protected Clock clock;

lib/src/main/java/com/google/cloud/datalineage/producerclient/LineageEnablementCache.java

@@ -16,32 +16,35 @@
 
 import java.time.Duration;
 
-/** Cache used to store information about whether the API is disabled for a given project. */
+/**
+ * Cache used to store information about whether Lineage is disabled for a given project in
+ * LineageConfigurations.
+ */
 public interface LineageEnablementCache {
 
   /**
-   * Mark service state as disabled for a given project name.
+   * Mark lineage ingestion state as disabled for a given project name.
    *
-   * @param projectName The project for which to disable the service
+   * @param projectName The project for which to disable the lineage ingestion
    * @see LineageEnablementCache#markLineageAsDisabled(String, Duration)
    */
   void markLineageAsDisabled(String projectName);
 
   /**
-   * Mark service state as disabled for a given project name and duration.
+   * Mark lineage ingestion state as disabled for a given project name and duration.
    *
-   * @param projectName The project for which to disable the service
+   * @param projectName The project for which to disable the lineage ingestion
    * @param duration - suggests how long the project should be marked as disabled. It is not
-   *     guarantied that cache will indicate service state as disabled for given time. Behaviour
-   *     depends on implementation.
+   *     guarantied that cache will indicate lineage ingestion enablement state as disabled for
+   *     given time. Behaviour depends on implementation.
    */
   void markLineageAsDisabled(String projectName, Duration duration);
 
   /**
-   * Indicates if service with provided projectName is marked as disabled.
+   * Indicates if lineage ingestion with provided projectName is marked as disabled.
    *
-   * @param projectName The project for which to disable the service
-   * @return `true` is the service is marked as disabled
+   * @param projectName The project for which to disable the lineage ingestion
+   * @return `true` if the lineage ingestion is marked as disabled
    */
   boolean isLineageMarkedAsDisabled(String projectName);
 }

lib/src/main/java/com/google/cloud/datalineage/producerclient/LineageEnablementCacheFactory.java

@@ -1,3 +1,17 @@
+// Copyright 2024 Google LLC
+//
+// Licensed 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
+//
+// https://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 com.google.cloud.datalineage.producerclient;
 
 /** A factory that returns ConnectionCache based on LineageEnablementCacheSettings. */

lib/src/main/java/com/google/cloud/datalineage/producerclient/NoOpLineageEnablementCache.java

@@ -16,7 +16,7 @@
 
 import java.time.Duration;
 
-/** No-op implementation of the ApiEnablementCache that does nothing. */
+/** No-op implementation of the LineageEnablementCache that does nothing. */
 public class NoOpLineageEnablementCache implements LineageEnablementCache {
 
   @Override

lib/src/main/java/com/google/cloud/datalineage/producerclient/ProjectStatusCache.java

@@ -24,10 +24,11 @@
 /**
  * Generic cache to indicate whether a feature is disabled for a given project.
  *
- * <p>This class is thread-safe.
+ * <p> This class is thread-safe. There is no eviction guaranteed in case of cache overload.
  */
 @Slf4j
 public class ProjectStatusCache {
+
   private final Cache<String, LocalDateTime> projectToLockEndTime;
   private final Duration defaultCacheDisabledStatusTime;
   private final Clock clock;
@@ -41,7 +42,8 @@ public ProjectStatusCache(CacheOptions options, String cacheName) {
         options.getDefaultCacheDisabledStatusTime());
     this.defaultCacheDisabledStatusTime = options.getDefaultCacheDisabledStatusTime();
     this.clock = options.getClock();
-    this.projectToLockEndTime = CacheBuilder.newBuilder().maximumSize(options.getCacheSize()).build();
+    this.projectToLockEndTime = CacheBuilder.newBuilder().maximumSize(options.getCacheSize())
+        .build();
     this.cacheName = cacheName;
   }
 

lib/src/main/java/com/google/cloud/datalineage/producerclient/StandardLineageEnablementCache.java

@@ -17,11 +17,12 @@
 import java.time.Duration;
 
 /**
- * Cache used to indicate whether Lineage is disabled for a given project.
+ * Cache used to indicate whether Lineage Ingestion is disabled for a given project.
  *
  * <p>This class is a wrapper around a generic ProjectStatusCache.
  */
 public class StandardLineageEnablementCache implements LineageEnablementCache {
+
   private final ProjectStatusCache delegate;
 
   StandardLineageEnablementCache(CacheOptions options) {

lib/src/main/java/com/google/cloud/datalineage/producerclient/helpers/GrpcHelper.java

@@ -16,11 +16,14 @@
 
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.StatusCode.Code;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import com.google.protobuf.Any;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.rpc.ErrorInfo;
 import com.google.rpc.Status;
 import io.grpc.protobuf.StatusProto;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -36,18 +39,18 @@ private GrpcHelper() {
   }
 
   /**
-   * Gets reason field from grpc response. ets reason field from grpc response.
+   * Returns a set of error reasons from <code>com.google.rpc.Status</code> of a gRPC Exception.
    *
    * @param grpcException - error returned form grpc call
-   * @return string with value from reason field and null if it is unable to extract reason
+   * @return a set of strings with reasons extracted from Status if any
    */
-  public static String getReason(Throwable grpcException) {
-    log.debug("Extracting reason from gRPC exception: {}", grpcException.getMessage());
+  public static ImmutableSet<String> getErrorReasons(Throwable grpcException) {
+    log.debug("Extracting reasons from gRPC exception: {}", grpcException.getMessage());
     Status statusProto = StatusProto.fromThrowable(grpcException);
     if (statusProto == null) {
-      log.debug(
+      log.error(
           "Provided throwable is not a gRPC exception: {}", grpcException.getClass().getName());
-      return null;
+      throw new IllegalArgumentException("Provided throwable is not a gRPC exception");
     }
     /* Status is a standard way to represent API error.
      * This model consists of code, message and details.
@@ -57,20 +60,18 @@ public static String getReason(Throwable grpcException) {
      * - user can introduce new ones.
      * That's why in order to get ErrorInfo, we need to iterate over details and check all of them.
      */
-    for (Any any : statusProto.getDetailsList()) {
-      if (any.is(ErrorInfo.class)) {
-        try {
-          String reason = any.unpack(ErrorInfo.class).getReason();
-          log.debug("Successfully extracted reason from ErrorInfo: {}", reason);
-          return reason;
-        } catch (InvalidProtocolBufferException exception) {
-          log.warn("Invalid protocol buffer message while extracting ErrorInfo", exception);
-          return null;
-        }
-      }
-    }
-    log.debug("Message does not contain ErrorInfo for exception: {}", grpcException.getMessage());
-    return null;
+    return statusProto.getDetailsList().stream().filter((detail) -> detail.is(ErrorInfo.class))
+        .map((errorInfo) -> {
+          try {
+            String reason = errorInfo.unpack(
+                ErrorInfo.class).getReason();
+            log.debug("Successfully extracted reason from ErrorInfo: {}", reason);
+            return reason;
+          } catch (InvalidProtocolBufferException e) {
+            log.error("Invalid protocol buffer message while extracting ErrorInfo", e);
+            throw new IllegalArgumentException("Invalid protocol buffer message", e);
+          }
+        }).collect(ImmutableSet.toImmutableSet());
   }
 
   /**