Skip to content

Commit fd92dbe

Browse files
committed
Integrate App Engine Task Queue API with Google Cloud Tasks v2 for GA.
Routes App Engine push queue operations to Google Cloud Tasks v2 when the APPENGINE_USE_CLOUDTASK_PUSH_QUEUE environment variable is enabled, while retaining v2beta3 for QueueStatistics. Key changes include: - Adding CloudTasksClientWrapper to handle task creation, batch creation, deletion, batch deletion, and queue purging using the Cloud Tasks v2 gRPC client SDK (with v2beta3 retained for QueueStatistics). - Implementing transactional task queuing via a Datastore-backed buffer (_AE_PendingCloudTask) and a background sweeper (SweeperServlet and TaskProcessor) using the Cloud Tasks v2 gRPC client SDK. - Updating QueueImpl and TaskHandle to delegate push queue operations to the Cloud Tasks integration. - Updating Maven and Bazel build files to include the required Google Cloud Tasks, GAX, and proto dependencies. PiperOrigin-RevId: 999953510 Change-Id: I17726a56170136cb7fcdc11949c8555ac62f9ca1
1 parent aefbe4a commit fd92dbe

18 files changed

Lines changed: 5205 additions & 57 deletions

File tree

‎api/pom.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,10 @@
122122
<groupId>javax.servlet</groupId>
123123
<artifactId>javax.servlet-api</artifactId>
124124
</dependency>
125+
<dependency>
126+
<groupId>com.google.cloud</groupId>
127+
<artifactId>google-cloud-tasks</artifactId>
128+
</dependency>
125129
<dependency>
126130
<groupId>org.antlr</groupId>
127131
<artifactId>antlr-runtime</artifactId>

‎api/src/main/java/com/google/appengine/api/datastore/DatastoreApiHelper.java‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,9 @@
3131
import com.google.rpc.Code;
3232
import java.util.ConcurrentModificationException;
3333
import java.util.List;
34+
import java.util.Objects;
3435
import java.util.concurrent.Future;
36+
import org.jspecify.annotations.Nullable;
3537

3638
/**
3739
* Helper methods and constants shared by classes that implement the Java api on top of the
@@ -56,6 +58,64 @@ public final class DatastoreApiHelper {
5658

5759
private DatastoreApiHelper() {}
5860

61+
/**
62+
* Registers a callback to be executed immediately after the specified transaction commits. If the
63+
* transaction is rolled back or fails to commit, the callback will not be invoked.
64+
*
65+
* <p>Transactions wrapped by frameworks (for example Objectify) are resolved to the native
66+
* transaction active on the current thread with the same app and transaction ID.
67+
*
68+
* @return {@code true} if the callback was registered, {@code false} if the native transaction
69+
* could not be resolved
70+
*/
71+
public static boolean addPostCommitCallback(Transaction txn, Runnable callback) {
72+
if (callback == null) {
73+
return false;
74+
}
75+
TransactionImpl txnImpl = resolveTransactionImpl(txn);
76+
if (txnImpl == null) {
77+
return false;
78+
}
79+
txnImpl.addPostCommitCallback(callback);
80+
return true;
81+
}
82+
83+
/**
84+
* Returns the native transaction backing {@code txn}, or {@code txn} itself if it cannot be
85+
* resolved.
86+
*/
87+
public static Transaction unwrapTransaction(Transaction txn) {
88+
TransactionImpl txnImpl = resolveTransactionImpl(txn);
89+
return txnImpl != null ? txnImpl : txn;
90+
}
91+
92+
private static @Nullable TransactionImpl resolveTransactionImpl(@Nullable Transaction txn) {
93+
if (txn == null) {
94+
return null;
95+
}
96+
if (txn instanceof TransactionImpl txnImpl) {
97+
return txnImpl;
98+
}
99+
try {
100+
String id = txn.getId();
101+
String app = txn.getApp();
102+
if (id == null) {
103+
return null;
104+
}
105+
for (Transaction active :
106+
DatastoreServiceFactory.getDatastoreService().getActiveTransactions()) {
107+
if (active instanceof TransactionImpl activeImpl
108+
&& id.equals(activeImpl.getId())
109+
&& Objects.equals(app, activeImpl.getApp())) {
110+
return activeImpl;
111+
}
112+
}
113+
} catch (RuntimeException e) {
114+
// Wrapped transaction could not be resolved; fall through.
115+
}
116+
return null;
117+
}
118+
59119
// Corresponds to _ToDatastoreError in datastore.py.
60120
// Keep in sync!
61121
public static RuntimeException translateError(ApiProxy.ApplicationException exception) {

‎api/src/main/java/com/google/appengine/api/datastore/TransactionImpl.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,13 @@ enum TransactionState {
8181

8282
TransactionState state = TransactionState.BEGUN;
8383

84+
private final List<Runnable> postCommitCallbacks =
85+
new java.util.concurrent.CopyOnWriteArrayList<>();
86+
87+
void addPostCommitCallback(Runnable callback) {
88+
postCommitCallbacks.add(callback);
89+
}
90+
8491
/** A {@link PostOpFuture} implementation that runs both post put and post delete callbacks. */
8592
private class PostCommitFuture extends PostOpFuture<Void> {
8693
private final List<Entity> putEntities;
@@ -100,6 +107,13 @@ void executeCallbacks(Void ignoreMe) {
100107
DeleteContext deleteContext =
101108
new DeleteContext(TransactionImpl.this, TransactionImpl.this, deletedKeys);
102109
callbacks.executePostDeleteCallbacks(deleteContext);
110+
for (Runnable callback : postCommitCallbacks) {
111+
try {
112+
callback.run();
113+
} catch (Throwable t) {
114+
logger.log(Level.SEVERE, "Exception in post-commit callback: " + t.getMessage(), t);
115+
}
116+
}
103117
}
104118
}
105119

0 commit comments

Comments
 (0)