CDAP-21261 : Lease or Locking support for Secure Store - RTR Oauth

cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStore.java

@@ -18,7 +18,10 @@
 
 import io.cdap.cdap.api.annotation.Beta;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
+import java.util.Set;
+
 
 /**
  * Provides read access to the secure store. For write access use {@link SecureStoreManager}.
@@ -73,4 +76,13 @@ default SecureStoreMetadata getMetadata(String namespace, String name) throws Ex
   default byte[] getData(String namespace, String name) throws Exception {
     return get(namespace, name).get();
   }
+
+  /**
+   * Returns the capabilities supported by this secure store implementation.
+   *
+   * @return A set of supported capabilities
+   */
+  default Set<SecureStoreCapability> getCapabilities() throws IOException {
+    return Collections.emptySet();
+  }
 }

cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStoreCapability.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * 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
+ *
+ * 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 io.cdap.cdap.api.security.store;
+
+/**
+ * Represents the capabilities of the Secure Store.
+ */
+public enum SecureStoreCapability {
+  LEASE
+}

cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStoreLease.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * 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
+ *
+ * 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 io.cdap.cdap.api.security.store;
+
+import java.util.Objects;
+
+/**
+ * API model representing a distributed lease lock on a secret or credential update in SecureStore.
+ * <p>
+ * Note: This is the caller-facing API counterpart to {@code io.cdap.cdap.securestore.spi.lease.SecretLease}
+ * in the SPI layer, mirroring the {@link io.cdap.cdap.api.security.store.SecureStoreMetadata} vs {@code SecretMetadata}
+ * architectural pattern in CDAP.
+ */
+public class SecureStoreLease {
+  private final boolean acquired;
+  private final String lockTimestamp;
+  private final String lockHolder;
+
+  private SecureStoreLease(boolean acquired, String lockTimestamp, String lockHolder) {
+    this.acquired = acquired;
+    this.lockTimestamp = lockTimestamp;
+    this.lockHolder = lockHolder;
+  }
+
+  public static SecureStoreLease acquired(String lockTimestamp, String lockHolder) {
+    return new SecureStoreLease(true, lockTimestamp, lockHolder);
+  }
+
+  public static SecureStoreLease failed() {
+    return new SecureStoreLease(false, null, null);
+  }
+
+  public boolean isAcquired() {
+    return acquired;
+  }
+
+  public String getLockTimestamp() {
+    return lockTimestamp;
+  }
+
+  public String getLockHolder() {
+    return lockHolder;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    SecureStoreLease that = (SecureStoreLease) o;
+    return acquired == that.acquired
+        && Objects.equals(lockTimestamp, that.lockTimestamp)
+        && Objects.equals(lockHolder, that.lockHolder);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(acquired, lockTimestamp, lockHolder);
+  }
+
+  @Override
+  public String toString() {
+    return "SecureStoreLease{"
+        + "acquired=" + acquired
+        + ", lockTimestamp='" + lockTimestamp + '\''
+        + ", lockHolder='" + lockHolder + '\''
+        + '}';
+  }
+}

cdap-api/src/main/java/io/cdap/cdap/api/security/store/SecureStoreManager.java

@@ -67,4 +67,31 @@
    * @throws Exception If the specified namespace or name does not exist
    */
   void delete(String namespace, String name) throws Exception;
+
+
+  /**
+   * Attempts to acquire a lease lock on a secret resource.
+   *
+   * @param namespace The namespace that this key belongs to
+   * @param name Name of the secure key
+   * @param timeoutMs Lock timeout in milliseconds before lease is considered expired
+   * @return {@link SecureStoreLease} indicating acquisition success and lock details
+   * @throws Exception If lock acquisition fails due to underlying storage errors
+   */
+  default SecureStoreLease acquireLease(String namespace, String name, long timeoutMs,
+                                          String lockHolder) throws Exception {
+    throw new UnsupportedOperationException("Leases are not supported by this SecureStore implementation.");
+  }
+
+  /**
+   * Releases an acquired lease lock on a secret resource.
+   *
+   * @param namespace The namespace that this key belongs to
+   * @param name Name of the secure key
+   * @param lease {@link SecureStoreLease} to release
+   * @throws Exception If lock release fails due to underlying storage errors
+   */
+  default void releaseLease(String namespace, String name, SecureStoreLease lease) throws Exception {
+    throw new UnsupportedOperationException("Leases are not supported by this SecureStore implementation.");
+  }
 }

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/AbstractContext.java

@@ -59,6 +59,7 @@
 import io.cdap.cdap.api.schedule.TriggeringScheduleInfo;
 import io.cdap.cdap.api.security.AccessException;
 import io.cdap.cdap.api.security.store.SecureStore;
+import io.cdap.cdap.api.security.store.SecureStoreCapability;
 import io.cdap.cdap.api.security.store.SecureStoreData;
 import io.cdap.cdap.api.security.store.SecureStoreManager;
 import io.cdap.cdap.api.security.store.SecureStoreMetadata;
@@ -99,8 +100,8 @@
 import io.cdap.cdap.internal.app.runtime.schedule.TriggeringScheduleInfoAdapter;
 import io.cdap.cdap.logging.context.LoggingContextHelper;
 import io.cdap.cdap.messaging.spi.MessagingService;
 import io.cdap.cdap.messaging.context.BasicMessagingAdmin;
 import io.cdap.cdap.messaging.context.MultiThreadMessagingContext;
 import io.cdap.cdap.proto.Notification;
 import io.cdap.cdap.proto.id.ApplicationId;
 import io.cdap.cdap.proto.id.ArtifactId;
@@ -132,7 +133,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * Base class for program runtime context
  */
 public abstract class AbstractContext extends AbstractServiceDiscoverer
@@ -535,6 +536,12 @@
     return Retries.callWithRetries(() -> secureStore.getData(namespace, name), retryStrategy);
   }
 
+
+  @Override
+  public Set<SecureStoreCapability> getCapabilities() throws java.io.IOException {
+    return secureStore.getCapabilities();
+  }
+
   @Override
   public void execute(final TxRunnable runnable) throws TransactionFailureException {
     execute(runnable, false);
@@ -591,7 +598,7 @@
         .getDataTracer(applicationId, dataTracerName);
   }
 
   @Nullable
   @Override
   public TriggeringScheduleInfo getTriggeringScheduleInfo() {
     return triggeringScheduleInfo;
@@ -601,7 +608,7 @@
    * Run some code with the context class loader combined from the program class loader and the
    * system class loader.
    */
   public void execute(ThrowingRunnable runnable) throws Exception {
     ClassLoader oldClassLoader = ClassLoaders.setContextClassLoader(
         getProgramInvocationClassLoader());
     try {
@@ -765,7 +772,7 @@
     return messagingContext;
   }
 
   @Override
   public Map<MetadataScope, Metadata> getMetadata(MetadataEntity metadataEntity)
       throws MetadataException {
     return Retries.callWithRetries(() -> metadataReader.getMetadata(metadataEntity), retryStrategy);
@@ -842,7 +849,7 @@
         retryStrategy);
   }
 
   /**
    * @return Map with feature flags defined in CConf.
    */
   @Override
@@ -913,7 +920,7 @@
     };
   }
 
   /**
    * @return the {@link Set} of field lineage operations
    */
   public Set<Operation> getFieldLineageOperations() {

cdap-app-fabric/src/main/java/io/cdap/cdap/internal/app/runtime/DefaultAdmin.java

@@ -21,6 +21,7 @@
 import io.cdap.cdap.api.messaging.MessagingAdmin;
 import io.cdap.cdap.api.messaging.TopicAlreadyExistsException;
 import io.cdap.cdap.api.messaging.TopicNotFoundException;
+import io.cdap.cdap.api.security.store.SecureStoreLease;
 import io.cdap.cdap.api.security.store.SecureStoreManager;
 import io.cdap.cdap.common.NamespaceNotFoundException;
 import io.cdap.cdap.common.namespace.NamespaceQueryAdmin;
@@ -91,6 +92,19 @@ public void delete(String namespace, String name) throws Exception {
     Retries.runWithRetries(() -> secureStoreManager.delete(namespace, name), retryStrategy);
   }
 
+
+  @Override
+  public SecureStoreLease acquireLease(String namespace, String name, long timeoutMs,
+                                       String lockHolder) throws Exception {
+    return Retries.callWithRetries(
+        () -> secureStoreManager.acquireLease(namespace, name, timeoutMs, lockHolder), retryStrategy);
+  }
+
+  @Override
+  public void releaseLease(String namespace, String name, SecureStoreLease lease) throws Exception {
+    Retries.runWithRetries(() -> secureStoreManager.releaseLease(namespace, name, lease), retryStrategy);
+  }
+
   @Override
   public void createTopic(final String topic) throws TopicAlreadyExistsException, IOException {
     if (messagingAdmin == null) {

cdap-securestore-ext-gcp-secretstore/src/main/java/io/cdap/cdap/securestore/gcp/cloudsecretmanager/CloudSecretManagerClient.java

@@ -19,6 +19,7 @@
 import com.google.api.gax.core.CredentialsProvider;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.ServiceOptions;
 import com.google.cloud.secretmanager.v1.AddSecretVersionRequest;
@@ -35,15 +36,15 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.Map;
 import java.util.Optional;
 
 /** Client for <a href="https://cloud.google.com/secret-manager">Google Cloud Secret Manager</a> */
 public class CloudSecretManagerClient {
   private static final Logger LOG = LoggerFactory.getLogger(CloudSecretManagerClient.class);
   /**
@@ -175,6 +176,32 @@
       fieldMask.build());
   }
 
+  /**
+   * Conditionally updates annotations on the specified secret using an ETag for optimistic concurrency control.
+   *
+   * @param namespace CDAP secret namespace
+   * @param name CDAP secret name
+   * @param annotationsToUpdate Map of annotations to update
+   * @param etag The expected ETag of the secret
+   */
+  public void updateSecretAnnotations(String namespace,
+                                     String name,
+                                     Map<String, String> annotationsToUpdate,
+                                     String etag) {
+    String resourceName = getSecretResourceName(namespace, name);
+    Secret.Builder secretBuilder = Secret.newBuilder()
+        .setName(resourceName)
+        .setEtag(etag);
+
+    for (Map.Entry<String, String> entry : annotationsToUpdate.entrySet()) {
+      secretBuilder.putAnnotations(entry.getKey(), entry.getValue());
+    }
+
+    secretManager.updateSecret(
+        secretBuilder.build(),
+        FieldMask.newBuilder().addPaths("annotations").build());
+  }
+
   /**
    * Deletes the specified secret.
    *