Use the replica jpaTm in FKI and EppResource cache methods (#1503)

The cached methods are only used in situations where we don't really
care about being 100% synchronously up to date (e.g. whois), and they're
not used frequently anyway, so it's safe to use the replica in these
locations.

core/build.gradle

@@ -676,9 +676,9 @@ Optional<List<String>> getToolArgsList() {
 
 // To run the nomulus tools with these command line tokens:
 // "--foo", "bar baz", "--qux=quz"
-//   gradle registryTool --args="--foo 'bar baz' --qux=quz"
+//   gradle core:registryTool --args="--foo 'bar baz' --qux=quz"
 // or:
-//   gradle registryTool --PtoolArgs="--foo|bar baz|--qux=quz"
+//   gradle core:registryTool -PtoolArgs="--foo|bar baz|--qux=quz"
 // Note that the delimiting pipe can be backslash escaped if it is part of a
 // parameter.
 ext.createToolTask = {
@@ -708,6 +708,9 @@ createToolTask(
 createToolTask(
     'validateSqlPipeline', 'google.registry.beam.comparedb.ValidateSqlPipeline')
 
+createToolTask(
+    'validateDatastorePipeline', 'google.registry.beam.comparedb.ValidateDatastorePipeline')
+
 
 createToolTask(
     'jpaDemoPipeline', 'google.registry.beam.common.JpaDemoPipeline')
@@ -793,6 +796,16 @@ if (environment == 'alpha') {
               mainClass: 'google.registry.beam.rde.RdePipeline',
               metaData : 'google/registry/beam/rde_pipeline_metadata.json'
           ],
+      validateDatastore  :
+          [
+              mainClass: 'google.registry.beam.comparedb.ValidateDatastorePipeline',
+              metaData: 'google/registry/beam/validate_datastore_pipeline_metadata.json'
+          ],
+      validateSql  :
+          [
+              mainClass: 'google.registry.beam.comparedb.ValidateSqlPipeline',
+              metaData: 'google/registry/beam/validate_sql_pipeline_metadata.json'
+          ],
   ]
   project.tasks.create("stageBeamPipelines") {
     doLast {

core/src/main/java/google/registry/backup/CommitLogCheckpointAction.java

@@ -17,7 +17,7 @@ package google.registry.backup;
 import static google.registry.backup.ExportCommitLogDiffAction.LOWER_CHECKPOINT_TIME_PARAM;
 import static google.registry.backup.ExportCommitLogDiffAction.UPPER_CHECKPOINT_TIME_PARAM;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 
 import com.google.common.collect.ImmutableMultimap;
@@ -67,7 +67,8 @@ public final class CommitLogCheckpointAction implements Runnable {
     final CommitLogCheckpoint checkpoint = strategy.computeCheckpoint();
     logger.atInfo().log(
         "Generated candidate checkpoint for time: %s", checkpoint.getCheckpointTime());
-    tm().transact(
+    ofyTm()
+        .transact(
             () -> {
               DateTime lastWrittenTime = CommitLogCheckpointRoot.loadRoot().getLastWrittenTime();
               if (isBeforeOrAt(checkpoint.getCheckpointTime(), lastWrittenTime)) {

core/src/main/java/google/registry/backup/DeleteOldCommitLogsAction.java

@@ -18,7 +18,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static google.registry.mapreduce.MapreduceRunner.PARAM_DRY_RUN;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static java.lang.Boolean.FALSE;
 import static java.lang.Boolean.TRUE;
 
@@ -288,7 +288,8 @@ public final class DeleteOldCommitLogsAction implements Runnable {
       }
 
       DeletionResult deletionResult =
-          tm().transactNew(
+          ofyTm()
+              .transactNew(
                   () -> {
                     CommitLogManifest manifest = auditedOfy().load().key(manifestKey).now();
                     // It is possible that the same manifestKey was run twice, if a shard had to be

core/src/main/java/google/registry/batch/AsyncTaskEnqueuer.java

@@ -110,7 +110,7 @@ public final class AsyncTaskEnqueuer {
             .method(Method.POST)
             .header("Host", backendHostname)
             .countdownMillis(etaDuration.getMillis())
-            .param(PARAM_RESOURCE_KEY, entityKey.getOfyKey().getString())
+            .param(PARAM_RESOURCE_KEY, entityKey.stringify())
             .param(PARAM_REQUESTED_TIME, now.toString());
     if (whenToResave.size() > 1) {
       task.param(PARAM_RESAVE_TIMES, Joiner.on(',').join(whenToResave.tailSet(firstResave, false)));
@@ -131,7 +131,7 @@ public final class AsyncTaskEnqueuer {
     TaskOptions task =
         TaskOptions.Builder.withMethod(Method.PULL)
             .countdownMillis(asyncDeleteDelay.getMillis())
-            .param(PARAM_RESOURCE_KEY, resourceToDelete.createVKey().getOfyKey().getString())
+            .param(PARAM_RESOURCE_KEY, resourceToDelete.createVKey().stringify())
             .param(PARAM_REQUESTING_CLIENT_ID, requestingRegistrarId)
             .param(PARAM_SERVER_TRANSACTION_ID, trid.getServerTransactionId())
             .param(PARAM_IS_SUPERUSER, Boolean.toString(isSuperuser))
@@ -148,7 +148,7 @@ public final class AsyncTaskEnqueuer {
     addTaskToQueueWithRetry(
         asyncDnsRefreshPullQueue,
         TaskOptions.Builder.withMethod(Method.PULL)
-            .param(PARAM_HOST_KEY, hostKey.getOfyKey().getString())
+            .param(PARAM_HOST_KEY, hostKey.stringify())
             .param(PARAM_REQUESTED_TIME, now.toString()));
   }
 

core/src/main/java/google/registry/beam/common/RegistryPipelineComponent.java

@@ -23,6 +23,7 @@ import google.registry.config.RegistryConfig.ConfigModule;
 import google.registry.persistence.PersistenceModule;
 import google.registry.persistence.PersistenceModule.BeamBulkQueryJpaTm;
 import google.registry.persistence.PersistenceModule.BeamJpaTm;
+import google.registry.persistence.PersistenceModule.BeamReadOnlyReplicaJpaTm;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.privileges.secretmanager.SecretManagerModule;
@@ -59,6 +60,13 @@ public interface RegistryPipelineComponent {
   @BeamBulkQueryJpaTm
   Lazy<JpaTransactionManager> getBulkQueryJpaTransactionManager();
 
+  /**
+   * A {@link JpaTransactionManager} that uses the Postgres read-only replica if configured (uses
+   * the standard DB otherwise).
+   */
+  @BeamReadOnlyReplicaJpaTm
+  Lazy<JpaTransactionManager> getReadOnlyReplicaJpaTransactionManager();
+
   @Component.Builder
   interface Builder {
 

core/src/main/java/google/registry/beam/common/RegistryPipelineWorkerInitializer.java

@@ -56,6 +56,10 @@ public class RegistryPipelineWorkerInitializer implements JvmInitializer {
       case BULK_QUERY:
         transactionManagerLazy = registryPipelineComponent.getBulkQueryJpaTransactionManager();
         break;
+      case READ_ONLY_REPLICA:
+        transactionManagerLazy =
+            registryPipelineComponent.getReadOnlyReplicaJpaTransactionManager();
+        break;
       case REGULAR:
       default:
         transactionManagerLazy = registryPipelineComponent.getJpaTransactionManager();

core/src/main/java/google/registry/beam/comparedb/DatastoreSnapshots.java

@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
 import google.registry.backup.VersionedEntity;
 import google.registry.beam.initsql.Transforms;
+import google.registry.model.EppResource;
 import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.common.Cursor;
@@ -42,6 +43,7 @@ import google.registry.model.tld.Registry;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.ParDo;
@@ -93,7 +95,8 @@ public final class DatastoreSnapshots {
       String commitLogDir,
       DateTime commitLogFromTime,
       DateTime commitLogToTime,
-      Set<Class<?>> kinds) {
+      Set<Class<?>> kinds,
+      Optional<DateTime> compareStartTime) {
     PCollectionTuple snapshot =
         pipeline.apply(
             "Load Datastore snapshot.",
@@ -112,11 +115,11 @@ public final class DatastoreSnapshots {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag((Class<? extends SqlEntity>) kind),
-                datastoreEntityToPojo(perKindSnapshot, kind.getSimpleName()));
+                datastoreEntityToPojo(perKindSnapshot, kind.getSimpleName(), compareStartTime));
         continue;
       }
       Verify.verify(kind == HistoryEntry.class, "Unexpected Non-SqlEntity class: %s", kind);
-      PCollectionTuple historyEntriesByType = splitHistoryEntry(perKindSnapshot);
+      PCollectionTuple historyEntriesByType = splitHistoryEntry(perKindSnapshot, compareStartTime);
       for (Map.Entry<TupleTag<?>, PCollection<?>> entry :
           historyEntriesByType.getAll().entrySet()) {
         perTypeSnapshots = perTypeSnapshots.and(entry.getKey().getId(), entry.getValue());
@@ -129,7 +132,9 @@ public final class DatastoreSnapshots {
    * Splits a {@link PCollection} of {@link HistoryEntry HistoryEntries} into three collections of
    * its child entities by type.
    */
-  static PCollectionTuple splitHistoryEntry(PCollection<VersionedEntity> historyEntries) {
+  static PCollectionTuple splitHistoryEntry(
+      PCollection<VersionedEntity> historyEntries, Optional<DateTime> compareStartTime) {
+    DateTime nullableStartTime = compareStartTime.orElse(null);
     return historyEntries.apply(
         "Split HistoryEntry by Resource Type",
         ParDo.of(
@@ -138,6 +143,7 @@ public final class DatastoreSnapshots {
                   public void processElement(
                       @Element VersionedEntity historyEntry, MultiOutputReceiver out) {
                     Optional.ofNullable(Transforms.convertVersionedEntityToSqlEntity(historyEntry))
+                        .filter(e -> isEntityIncludedForComparison(e, nullableStartTime))
                         .ifPresent(
                             sqlEntity ->
                                 out.get(createSqlEntityTupleTag(sqlEntity.getClass()))
@@ -155,7 +161,8 @@ public final class DatastoreSnapshots {
    * objects.
    */
   static PCollection<SqlEntity> datastoreEntityToPojo(
-      PCollection<VersionedEntity> entities, String desc) {
+      PCollection<VersionedEntity> entities, String desc, Optional<DateTime> compareStartTime) {
+    DateTime nullableStartTime = compareStartTime.orElse(null);
     return entities.apply(
         "Datastore Entity to Pojo " + desc,
         ParDo.of(
@@ -164,8 +171,23 @@ public final class DatastoreSnapshots {
               public void processElement(
                   @Element VersionedEntity entity, OutputReceiver<SqlEntity> out) {
                 Optional.ofNullable(Transforms.convertVersionedEntityToSqlEntity(entity))
+                    .filter(e -> isEntityIncludedForComparison(e, nullableStartTime))
                     .ifPresent(out::output);
               }
             }));
   }
+
+  static boolean isEntityIncludedForComparison(
+      SqlEntity entity, @Nullable DateTime compareStartTime) {
+    if (compareStartTime == null) {
+      return true;
+    }
+    if (entity instanceof HistoryEntry) {
+      return compareStartTime.isBefore(((HistoryEntry) entity).getModificationTime());
+    }
+    if (entity instanceof EppResource) {
+      return compareStartTime.isBefore(((EppResource) entity).getUpdateTimestamp().getTimestamp());
+    }
+    return true;
+  }
 }

core/src/main/java/google/registry/beam/comparedb/LatestDatastoreSnapshotFinder.java

@@ -53,11 +53,11 @@ public class LatestDatastoreSnapshotFinder {
   }
 
   /**
-   * Finds information of the most recent Datastore snapshot, including the GCS folder of the
-   * exported data files and the start and stop times of the export. The folder of the CommitLogs is
-   * also included in the return.
+   * Finds information of the most recent Datastore snapshot that ends strictly before {@code
+   * exportEndTimeUpperBound}, including the GCS folder of the exported data files and the start and
+   * stop times of the export. The folder of the CommitLogs is also included in the return.
    */
-  public DatastoreSnapshotInfo getSnapshotInfo() {
+  public DatastoreSnapshotInfo getSnapshotInfo(Instant exportEndTimeUpperBound) {
     String bucketName = RegistryConfig.getDatastoreBackupsBucket().substring("gs://".length());
     /**
      * Find the bucket-relative path to the overall metadata file of the last Datastore export.
@@ -65,7 +65,8 @@ public class LatestDatastoreSnapshotFinder {
      * return value is like
      * "2021-11-19T06:00:00_76493/2021-11-19T06:00:00_76493.overall_export_metadata".
      */
-    Optional<String> metaFilePathOptional = findMostRecentExportMetadataFile(bucketName, 2);
+    Optional<String> metaFilePathOptional =
+        findNewestExportMetadataFileBeforeTime(bucketName, exportEndTimeUpperBound, 2);
     if (!metaFilePathOptional.isPresent()) {
       throw new NoSuchElementException("No exports found over the past 2 days.");
     }
@@ -85,8 +86,9 @@ public class LatestDatastoreSnapshotFinder {
   }
 
   /**
-   * Finds the bucket-relative path of the overall export metadata file, in the given bucket,
-   * searching back up to {@code lookBackDays} days, including today.
+   * Finds the latest Datastore export that ends strictly before {@code endTimeUpperBound} and
+   * returns the bucket-relative path of the overall export metadata file, in the given bucket. The
+   * search goes back for up to {@code lookBackDays} days in time, including today.
    *
    * <p>The overall export metadata file is the last file created during a Datastore export. All
    * data has been exported by the creation time of this file. The name of this file, like that of
@@ -95,7 +97,8 @@ public class LatestDatastoreSnapshotFinder {
    * <p>An example return value: {@code
    * 2021-11-19T06:00:00_76493/2021-11-19T06:00:00_76493.overall_export_metadata}.
    */
-  private Optional<String> findMostRecentExportMetadataFile(String bucketName, int lookBackDays) {
+  private Optional<String> findNewestExportMetadataFileBeforeTime(
+      String bucketName, Instant endTimeUpperBound, int lookBackDays) {
     DateTime today = clock.nowUtc();
     for (int day = 0; day < lookBackDays; day++) {
       String dateString = today.minusDays(day).toString("yyyy-MM-dd");
@@ -107,7 +110,11 @@ public class LatestDatastoreSnapshotFinder {
                 .sorted(Comparator.<String>naturalOrder().reversed())
                 .findFirst();
         if (metaFilePath.isPresent()) {
-          return metaFilePath;
+          BlobInfo blobInfo = gcsUtils.getBlobInfo(BlobId.of(bucketName, metaFilePath.get()));
+          Instant exportEndTime = new Instant(blobInfo.getCreateTime());
+          if (exportEndTime.isBefore(endTimeUpperBound)) {
+            return metaFilePath;
+          }
         }
       } catch (IOException ioe) {
         throw new RuntimeException(ioe);

core/src/main/java/google/registry/beam/comparedb/SqlSnapshots.java

@@ -14,15 +14,22 @@
 
 package google.registry.beam.comparedb;
 
+import static com.google.common.base.Preconditions.checkState;
 import static google.registry.beam.comparedb.ValidateSqlUtils.createSqlEntityTupleTag;
 import static google.registry.beam.comparedb.ValidateSqlUtils.getMedianIdForHistoryTable;
 
+import com.google.auto.value.AutoValue;
+import com.google.common.base.Strings;
 import com.google.common.base.Verify;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSetMultimap;
 import com.google.common.collect.Streams;
 import google.registry.beam.common.RegistryJpaIO;
+import google.registry.beam.common.RegistryJpaIO.Read;
+import google.registry.model.EppResource;
+import google.registry.model.UpdateAutoTimestamp;
 import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.bulkquery.BulkQueryEntities;
@@ -50,8 +57,10 @@ import google.registry.model.replay.SqlEntity;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.tld.Registry;
 import google.registry.persistence.transaction.CriteriaQueryBuilder;
+import google.registry.util.DateTimeUtils;
 import java.io.Serializable;
 import java.util.Optional;
+import javax.persistence.Entity;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.Flatten;
@@ -65,6 +74,7 @@ import org.apache.beam.sdk.values.PCollectionList;
 import org.apache.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.sdk.values.TypeDescriptors;
+import org.joda.time.DateTime;
 
 /**
  * Utilities for loading SQL snapshots.
@@ -113,28 +123,48 @@ public final class SqlSnapshots {
   public static PCollectionTuple loadCloudSqlSnapshotByType(
       Pipeline pipeline,
       ImmutableSet<Class<? extends SqlEntity>> sqlEntityTypes,
-      Optional<String> snapshotId) {
+      Optional<String> snapshotId,
+      Optional<DateTime> compareStartTime) {
     PCollectionTuple perTypeSnapshots = PCollectionTuple.empty(pipeline);
     for (Class<? extends SqlEntity> clazz : sqlEntityTypes) {
       if (clazz == DomainBase.class) {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag(DomainBase.class),
-                loadAndAssembleDomainBase(pipeline, snapshotId));
+                loadAndAssembleDomainBase(pipeline, snapshotId, compareStartTime));
         continue;
       }
       if (clazz == DomainHistory.class) {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag(DomainHistory.class),
-                loadAndAssembleDomainHistory(pipeline, snapshotId));
+                loadAndAssembleDomainHistory(pipeline, snapshotId, compareStartTime));
         continue;
       }
       if (clazz == ContactHistory.class) {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag(ContactHistory.class),
-                loadContactHistory(pipeline, snapshotId));
+                loadContactHistory(pipeline, snapshotId, compareStartTime));
+        continue;
+      }
+      if (clazz == HostHistory.class) {
+        perTypeSnapshots =
+            perTypeSnapshots.and(
+                createSqlEntityTupleTag(HostHistory.class),
+                loadHostHistory(
+                    pipeline, snapshotId, compareStartTime.orElse(DateTimeUtils.START_OF_TIME)));
+        continue;
+      }
+      if (EppResource.class.isAssignableFrom(clazz) && compareStartTime.isPresent()) {
+        perTypeSnapshots =
+            perTypeSnapshots.and(
+                createSqlEntityTupleTag(clazz),
+                pipeline.apply(
+                    "SQL Load " + clazz.getSimpleName(),
+                    buildEppResourceQueryWithTimeFilter(
+                            clazz, SqlEntity.class, snapshotId, compareStartTime.get())
+                        .withSnapshot(snapshotId.orElse(null))));
         continue;
       }
       perTypeSnapshots =
@@ -155,20 +185,33 @@ public final class SqlSnapshots {
    * @see BulkQueryEntities
    */
   public static PCollection<SqlEntity> loadAndAssembleDomainBase(
-      Pipeline pipeline, Optional<String> snapshotId) {
+      Pipeline pipeline, Optional<String> snapshotId, Optional<DateTime> compareStartTime) {
     PCollection<KV<String, Serializable>> baseObjects =
-        readAllAndAssignKey(pipeline, DomainBaseLite.class, DomainBaseLite::getRepoId, snapshotId);
+        readAllAndAssignKey(
+            pipeline,
+            DomainBaseLite.class,
+            DomainBaseLite::getRepoId,
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> gracePeriods =
-        readAllAndAssignKey(pipeline, GracePeriod.class, GracePeriod::getDomainRepoId, snapshotId);
+        readAllAndAssignKey(
+            pipeline,
+            GracePeriod.class,
+            GracePeriod::getDomainRepoId,
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> delegationSigners =
         readAllAndAssignKey(
             pipeline,
             DelegationSignerData.class,
             DelegationSignerData::getDomainRepoId,
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> domainHosts =
-        readAllAndAssignKey(pipeline, DomainHost.class, DomainHost::getDomainRepoId, snapshotId);
+        readAllAndAssignKey(
+            pipeline, DomainHost.class, DomainHost::getDomainRepoId, snapshotId, compareStartTime);
 
+    DateTime nullableCompareStartTime = compareStartTime.orElse(null);
     return PCollectionList.of(
             ImmutableList.of(baseObjects, gracePeriods, delegationSigners, domainHosts))
         .apply("SQL Merge DomainBase parts", Flatten.pCollections())
@@ -184,6 +227,14 @@ public final class SqlSnapshots {
                     TypedClassifier partsByType = new TypedClassifier(kv.getValue());
                     ImmutableSet<DomainBaseLite> baseObjects =
                         partsByType.getAllOf(DomainBaseLite.class);
+                    if (nullableCompareStartTime != null) {
+                      Verify.verify(
+                          baseObjects.size() <= 1,
+                          "Found duplicate DomainBaseLite object per repoId: " + kv.getKey());
+                      if (baseObjects.isEmpty()) {
+                        return;
+                      }
+                    }
                     Verify.verify(
                         baseObjects.size() == 1,
                         "Expecting one DomainBaseLite object per repoId: " + kv.getKey());
@@ -205,16 +256,16 @@ public final class SqlSnapshots {
    * <p>This method uses two queries to load data in parallel. This is a performance optimization
    * specifically for the production database.
    */
-  static PCollection<SqlEntity> loadContactHistory(Pipeline pipeline, Optional<String> snapshotId) {
-    long medianId =
-        getMedianIdForHistoryTable("ContactHistory")
-            .orElseThrow(
-                () -> new IllegalStateException("Not a valid database: no ContactHistory."));
+  static PCollection<SqlEntity> loadContactHistory(
+      Pipeline pipeline, Optional<String> snapshotId, Optional<DateTime> compareStartTime) {
+    PartitionedQuery partitionedQuery =
+        buildPartitonedHistoryQuery(ContactHistory.class, compareStartTime);
     PCollection<SqlEntity> part1 =
         pipeline.apply(
             "SQL Load ContactHistory first half",
             RegistryJpaIO.read(
-                    String.format("select c from ContactHistory c where id <= %s", medianId),
+                    partitionedQuery.firstHalfQuery(),
+                    partitionedQuery.parameters(),
                     false,
                     SqlEntity.class::cast)
                 .withSnapshot(snapshotId.orElse(null)));
@@ -222,7 +273,8 @@ public final class SqlSnapshots {
         pipeline.apply(
             "SQL Load ContactHistory second half",
             RegistryJpaIO.read(
-                    String.format("select c from ContactHistory c where id > %s", medianId),
+                    partitionedQuery.secondHalfQuery(),
+                    partitionedQuery.parameters(),
                     false,
                     SqlEntity.class::cast)
                 .withSnapshot(snapshotId.orElse(null)));
@@ -231,6 +283,19 @@ public final class SqlSnapshots {
         .apply("Combine ContactHistory parts", Flatten.pCollections());
   }
 
+  /** Loads all {@link HostHistory} entities from the database. */
+  static PCollection<SqlEntity> loadHostHistory(
+      Pipeline pipeline, Optional<String> snapshotId, DateTime compareStartTime) {
+    return pipeline.apply(
+        "SQL Load HostHistory",
+        RegistryJpaIO.read(
+                "select c from HostHistory c where :compareStartTime  < modificationTime",
+                ImmutableMap.of("compareStartTime", compareStartTime),
+                false,
+                SqlEntity.class::cast)
+            .withSnapshot(snapshotId.orElse(null)));
+  }
+
   /**
    * Bulk-loads all parts of {@link DomainHistory} and assembles them in the pipeline.
    *
@@ -240,16 +305,15 @@ public final class SqlSnapshots {
    * @see BulkQueryEntities
    */
   static PCollection<SqlEntity> loadAndAssembleDomainHistory(
-      Pipeline pipeline, Optional<String> snapshotId) {
-    long medianId =
-        getMedianIdForHistoryTable("DomainHistory")
-            .orElseThrow(
-                () -> new IllegalStateException("Not a valid database: no DomainHistory."));
+      Pipeline pipeline, Optional<String> snapshotId, Optional<DateTime> compareStartTime) {
+    PartitionedQuery partitionedQuery =
+        buildPartitonedHistoryQuery(DomainHistoryLite.class, compareStartTime);
     PCollection<KV<String, Serializable>> baseObjectsPart1 =
         queryAndAssignKey(
             pipeline,
             "first half",
-            String.format("select c from DomainHistory c where id <= %s", medianId),
+            partitionedQuery.firstHalfQuery(),
+            partitionedQuery.parameters(),
             DomainHistoryLite.class,
             compose(DomainHistoryLite::getDomainHistoryId, DomainHistoryId::toString),
             snapshotId);
@@ -257,7 +321,8 @@ public final class SqlSnapshots {
         queryAndAssignKey(
             pipeline,
             "second half",
-            String.format("select c from DomainHistory c where id > %s", medianId),
+            partitionedQuery.secondHalfQuery(),
+            partitionedQuery.parameters(),
             DomainHistoryLite.class,
             compose(DomainHistoryLite::getDomainHistoryId, DomainHistoryId::toString),
             snapshotId);
@@ -266,26 +331,31 @@ public final class SqlSnapshots {
             pipeline,
             GracePeriodHistory.class,
             compose(GracePeriodHistory::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> delegationSigners =
         readAllAndAssignKey(
             pipeline,
             DomainDsDataHistory.class,
             compose(DomainDsDataHistory::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> domainHosts =
         readAllAndAssignKey(
             pipeline,
             DomainHistoryHost.class,
             compose(DomainHistoryHost::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> transactionRecords =
         readAllAndAssignKey(
             pipeline,
             DomainTransactionRecord.class,
             compose(DomainTransactionRecord::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
 
+    DateTime nullableCompareStartTime = compareStartTime.orElse(null);
     return PCollectionList.of(
             ImmutableList.of(
                 baseObjectsPart1,
@@ -307,6 +377,15 @@ public final class SqlSnapshots {
                     TypedClassifier partsByType = new TypedClassifier(kv.getValue());
                     ImmutableSet<DomainHistoryLite> baseObjects =
                         partsByType.getAllOf(DomainHistoryLite.class);
+                    if (nullableCompareStartTime != null) {
+                      Verify.verify(
+                          baseObjects.size() <= 1,
+                          "Found duplicate DomainHistoryLite object per domainHistoryId: "
+                              + kv.getKey());
+                      if (baseObjects.isEmpty()) {
+                        return;
+                      }
+                    }
                     Verify.verify(
                         baseObjects.size() == 1,
                         "Expecting one DomainHistoryLite object per domainHistoryId: "
@@ -328,12 +407,19 @@ public final class SqlSnapshots {
       Pipeline pipeline,
       Class<R> type,
       SerializableFunction<R, String> keyFunction,
-      Optional<String> snapshotId) {
+      Optional<String> snapshotId,
+      Optional<DateTime> compareStartTime) {
+    Read<R, R> queryObject;
+    if (compareStartTime.isPresent() && EppResource.class.isAssignableFrom(type)) {
+      queryObject =
+          buildEppResourceQueryWithTimeFilter(type, type, snapshotId, compareStartTime.get());
+    } else {
+      queryObject =
+          RegistryJpaIO.read(() -> CriteriaQueryBuilder.create(type).build())
+              .withSnapshot(snapshotId.orElse(null));
+    }
     return pipeline
-        .apply(
-            "SQL Load " + type.getSimpleName(),
-            RegistryJpaIO.read(() -> CriteriaQueryBuilder.create(type).build())
-                .withSnapshot(snapshotId.orElse(null)))
+        .apply("SQL Load " + type.getSimpleName(), queryObject)
         .apply(
             "Assign Key to " + type.getSimpleName(),
             MapElements.into(
@@ -346,13 +432,15 @@ public final class SqlSnapshots {
       Pipeline pipeline,
       String diffrentiator,
       String jplQuery,
+      ImmutableMap<String, Object> queryParameters,
       Class<R> type,
       SerializableFunction<R, String> keyFunction,
       Optional<String> snapshotId) {
     return pipeline
         .apply(
             "SQL Load " + type.getSimpleName() + " " + diffrentiator,
-            RegistryJpaIO.read(jplQuery, false, type::cast).withSnapshot(snapshotId.orElse(null)))
+            RegistryJpaIO.read(jplQuery, queryParameters, false, type::cast)
+                .withSnapshot(snapshotId.orElse(null)))
         .apply(
             "Assign Key to " + type.getSimpleName() + " " + diffrentiator,
             MapElements.into(
@@ -367,6 +455,71 @@ public final class SqlSnapshots {
     return r -> f2.apply(f1.apply(r));
   }
 
+  static <R, T> Read<R, T> buildEppResourceQueryWithTimeFilter(
+      Class<R> entityType,
+      Class<T> castOutputAsType,
+      Optional<String> snapshotId,
+      DateTime compareStartTime) {
+    String tableName = getJpaEntityName(entityType);
+    String jpql =
+        String.format("select c from %s c where :compareStartTime < updateTimestamp", tableName);
+    return RegistryJpaIO.read(
+            jpql,
+            ImmutableMap.of("compareStartTime", UpdateAutoTimestamp.create(compareStartTime)),
+            false,
+            (R x) -> castOutputAsType.cast(x))
+        .withSnapshot(snapshotId.orElse(null));
+  }
+
+  static PartitionedQuery buildPartitonedHistoryQuery(
+      Class<?> entityType, Optional<DateTime> compareStartTime) {
+    String tableName = getJpaEntityName(entityType);
+    Verify.verify(
+        !Strings.isNullOrEmpty(tableName), "Invalid entity type %s", entityType.getSimpleName());
+    long medianId =
+        getMedianIdForHistoryTable(tableName)
+            .orElseThrow(() -> new IllegalStateException("Not a valid database: no " + tableName));
+    String firstHalfQuery = String.format("select c from %s c where id <= :historyId", tableName);
+    String secondHalfQuery = String.format("select c from %s c where id > :historyId", tableName);
+    if (compareStartTime.isPresent()) {
+      String timeFilter = "  and :compareStartTime  < modificationTime";
+      firstHalfQuery += timeFilter;
+      secondHalfQuery += timeFilter;
+      return PartitionedQuery.createPartitionedQuery(
+          firstHalfQuery,
+          secondHalfQuery,
+          ImmutableMap.of("historyId", medianId, "compareStartTime", compareStartTime.get()));
+    } else {
+      return PartitionedQuery.createPartitionedQuery(
+          firstHalfQuery, secondHalfQuery, ImmutableMap.of("historyId", medianId));
+    }
+  }
+
+  private static String getJpaEntityName(Class entityType) {
+    Entity entityAnnotation = (Entity) entityType.getAnnotation(Entity.class);
+    checkState(
+        entityAnnotation != null, "Unexpected non-entity type %s", entityType.getSimpleName());
+    return Strings.isNullOrEmpty(entityAnnotation.name())
+        ? entityType.getSimpleName()
+        : entityAnnotation.name();
+  }
+
+  /** Contains two queries that partition the target table in two. */
+  @AutoValue
+  abstract static class PartitionedQuery {
+    abstract String firstHalfQuery();
+
+    abstract String secondHalfQuery();
+
+    abstract ImmutableMap<String, Object> parameters();
+
+    public static PartitionedQuery createPartitionedQuery(
+        String firstHalfQuery, String secondHalfQuery, ImmutableMap<String, Object> parameters) {
+      return new AutoValue_SqlSnapshots_PartitionedQuery(
+          firstHalfQuery, secondHalfQuery, parameters);
+    }
+  }
+
   /** Container that receives mixed-typed data and groups them by {@link Class}. */
   static class TypedClassifier {
     private final ImmutableSetMultimap<Class<?>, Object> classifiedEntities;

core/src/main/java/google/registry/beam/comparedb/ValidateDatastorePipeline.java

@@ -0,0 +1,83 @@
+// Copyright 2022 The Nomulus Authors. All Rights Reserved.
+//
+// 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 google.registry.beam.comparedb;
+
+import google.registry.beam.common.RegistryPipelineOptions;
+import google.registry.beam.common.RegistryPipelineWorkerInitializer;
+import google.registry.beam.comparedb.LatestDatastoreSnapshotFinder.DatastoreSnapshotInfo;
+import google.registry.model.annotations.DeleteAfterMigration;
+import google.registry.persistence.PersistenceModule.JpaTransactionManagerType;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
+import java.util.Optional;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.joda.time.DateTime;
+
+/**
+ * Validates the asynchronous data replication process from Cloud SQL (primary) to Datastore
+ * (secondary).
+ *
+ * <p>This pipeline simply compares the snapshots provided by an invoker, which is responsible for
+ * obtaining two consistent snapshots for the same point of time.
+ */
+// TODO(weiminyu): Implement the invoker action in a followup PR.
+@DeleteAfterMigration
+public class ValidateDatastorePipeline {
+
+  private final ValidateDatastorePipelineOptions options;
+  private final LatestDatastoreSnapshotFinder datastoreSnapshotFinder;
+
+  public ValidateDatastorePipeline(
+      ValidateDatastorePipelineOptions options,
+      LatestDatastoreSnapshotFinder datastoreSnapshotFinder) {
+    this.options = options;
+    this.datastoreSnapshotFinder = datastoreSnapshotFinder;
+  }
+
+  void run(Pipeline pipeline) {
+    DateTime latestCommitLogTime = DateTime.parse(options.getLatestCommitLogTimestamp());
+    DatastoreSnapshotInfo mostRecentExport =
+        datastoreSnapshotFinder.getSnapshotInfo(latestCommitLogTime.toInstant());
+
+    ValidateSqlPipeline.setupPipeline(
+        pipeline,
+        Optional.ofNullable(options.getSqlSnapshotId()),
+        mostRecentExport,
+        latestCommitLogTime,
+        Optional.ofNullable(options.getComparisonStartTimestamp()).map(DateTime::parse));
+
+    pipeline.run();
+  }
+
+  public static void main(String[] args) {
+    ValidateDatastorePipelineOptions options =
+        PipelineOptionsFactory.fromArgs(args)
+            .withValidation()
+            .as(ValidateDatastorePipelineOptions.class);
+    RegistryPipelineOptions.validateRegistryPipelineOptions(options);
+
+    // Defensively set important options.
+    options.setIsolationOverride(TransactionIsolationLevel.TRANSACTION_REPEATABLE_READ);
+    options.setJpaTransactionManagerType(JpaTransactionManagerType.BULK_QUERY);
+
+    // Reuse Dataflow worker initialization code to set up JPA in the pipeline harness.
+    new RegistryPipelineWorkerInitializer().beforeProcessing(options);
+
+    LatestDatastoreSnapshotFinder datastoreSnapshotFinder =
+        DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
+            .datastoreSnapshotInfoFinder();
+    new ValidateDatastorePipeline(options, datastoreSnapshotFinder).run(Pipeline.create(options));
+  }
+}

core/src/main/java/google/registry/beam/comparedb/ValidateDatastorePipelineOptions.java

@@ -0,0 +1,39 @@
+// Copyright 2022 The Nomulus Authors. All Rights Reserved.
+//
+// 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 google.registry.beam.comparedb;
+
+import google.registry.model.annotations.DeleteAfterMigration;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.Validation;
+
+/** BEAM pipeline options for {@link ValidateDatastorePipelineOptions}. */
+@DeleteAfterMigration
+public interface ValidateDatastorePipelineOptions extends ValidateSqlPipelineOptions {
+
+  @Description(
+      "The id of the SQL snapshot to be compared with Datastore. "
+          + "If null, the current state of the SQL database is used.")
+  @Nullable
+  String getSqlSnapshotId();
+
+  void setSqlSnapshotId(String snapshotId);
+
+  @Description("The latest CommitLogs to load, in ISO8601 format.")
+  @Validation.Required
+  String getLatestCommitLogTimestamp();
+
+  void setLatestCommitLogTimestamp(String commitLogEndTimestamp);
+}

core/src/main/java/google/registry/beam/comparedb/ValidateSqlPipeline.java

@@ -26,6 +26,9 @@ import google.registry.beam.common.RegistryPipelineWorkerInitializer;
 import google.registry.beam.comparedb.LatestDatastoreSnapshotFinder.DatastoreSnapshotInfo;
 import google.registry.beam.comparedb.ValidateSqlUtils.CompareSqlEntity;
 import google.registry.model.annotations.DeleteAfterMigration;
+import google.registry.model.common.DatabaseMigrationStateSchedule;
+import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
+import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
 import google.registry.model.domain.DomainBase;
 import google.registry.model.domain.DomainHistory;
 import google.registry.model.replay.SqlEntity;
@@ -35,6 +38,7 @@ import google.registry.persistence.PersistenceModule.JpaTransactionManagerType;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.transaction.TransactionManagerFactory;
 import google.registry.util.RequestStatusChecker;
+import google.registry.util.SystemClock;
 import java.io.Serializable;
 import java.util.Optional;
 import org.apache.beam.sdk.Pipeline;
@@ -76,21 +80,16 @@ public class ValidateSqlPipeline {
       java.time.Duration.ofSeconds(30);
 
   private final ValidateSqlPipelineOptions options;
-  private final DatastoreSnapshotInfo mostRecentExport;
+  private final LatestDatastoreSnapshotFinder datastoreSnapshotFinder;
 
   public ValidateSqlPipeline(
-      ValidateSqlPipelineOptions options, DatastoreSnapshotInfo mostRecentExport) {
+      ValidateSqlPipelineOptions options, LatestDatastoreSnapshotFinder datastoreSnapshotFinder) {
     this.options = options;
-    this.mostRecentExport = mostRecentExport;
-  }
-
-  void run() {
-    run(Pipeline.create(options));
+    this.datastoreSnapshotFinder = datastoreSnapshotFinder;
   }
 
   @VisibleForTesting
   void run(Pipeline pipeline) {
-    // TODO(weiminyu): ensure migration stage is DATASTORE_PRIMARY or DATASTORE_PRIMARY_READ_ONLY
     Optional<Lock> lock = acquireCommitLogReplayLock();
     if (lock.isPresent()) {
       logger.atInfo().log("Acquired CommitLog Replay lock.");
@@ -101,6 +100,8 @@ public class ValidateSqlPipeline {
     try {
       DateTime latestCommitLogTime =
           TransactionManagerFactory.jpaTm().transact(() -> SqlReplayCheckpoint.get());
+      DatastoreSnapshotInfo mostRecentExport =
+          datastoreSnapshotFinder.getSnapshotInfo(latestCommitLogTime.toInstant());
       Preconditions.checkState(
           latestCommitLogTime.isAfter(mostRecentExport.exportInterval().getEnd()),
           "Cannot recreate Datastore snapshot since target time is in the middle of an export.");
@@ -109,7 +110,16 @@ public class ValidateSqlPipeline {
         lock.ifPresent(Lock::releaseSql);
         lock = Optional.empty();
 
-        setupPipeline(pipeline, Optional.of(databaseSnapshot.getSnapshotId()), latestCommitLogTime);
+        logger.atInfo().log(
+            "Starting comparison with export at %s and latestCommitLogTime at %s",
+            mostRecentExport.exportDir(), latestCommitLogTime);
+
+        setupPipeline(
+            pipeline,
+            Optional.of(databaseSnapshot.getSnapshotId()),
+            mostRecentExport,
+            latestCommitLogTime,
+            Optional.ofNullable(options.getComparisonStartTimestamp()).map(DateTime::parse));
         State state = pipeline.run().waitUntilFinish();
         if (!State.DONE.equals(state)) {
           throw new IllegalStateException("Unexpected pipeline state: " + state);
@@ -120,8 +130,12 @@ public class ValidateSqlPipeline {
     }
   }
 
-  void setupPipeline(
-      Pipeline pipeline, Optional<String> sqlSnapshotId, DateTime latestCommitLogTime) {
+  static void setupPipeline(
+      Pipeline pipeline,
+      Optional<String> sqlSnapshotId,
+      DatastoreSnapshotInfo mostRecentExport,
+      DateTime latestCommitLogTime,
+      Optional<DateTime> compareStartTime) {
     pipeline
         .getCoderRegistry()
         .registerCoderForClass(SqlEntity.class, SerializableCoder.of(Serializable.class));
@@ -135,11 +149,12 @@ public class ValidateSqlPipeline {
             // Increase by 1ms since we want to include commitLogs latestCommitLogTime but
             // this parameter is exclusive.
             latestCommitLogTime.plusMillis(1),
-            DatastoreSnapshots.ALL_DATASTORE_KINDS);
+            DatastoreSnapshots.ALL_DATASTORE_KINDS,
+            compareStartTime);
 
     PCollectionTuple cloudSqlSnapshot =
         SqlSnapshots.loadCloudSqlSnapshotByType(
-            pipeline, SqlSnapshots.ALL_SQL_ENTITIES, sqlSnapshotId);
+            pipeline, SqlSnapshots.ALL_SQL_ENTITIES, sqlSnapshotId, compareStartTime);
 
     verify(
         datastoreSnapshot.getAll().keySet().equals(cloudSqlSnapshot.getAll().keySet()),
@@ -212,6 +227,10 @@ public class ValidateSqlPipeline {
         return "ValidateSqlPipeline";
       }
 
+      LatestDatastoreSnapshotFinder datastoreSnapshotFinder =
+          DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
+              .datastoreSnapshotInfoFinder();
+
       @Override
       public boolean isRunning(String requestLogId) {
         return true;
@@ -230,11 +249,16 @@ public class ValidateSqlPipeline {
     // Reuse Dataflow worker initialization code to set up JPA in the pipeline harness.
     new RegistryPipelineWorkerInitializer().beforeProcessing(options);
 
-    DatastoreSnapshotInfo mostRecentExport =
-        DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
-            .datastoreSnapshotInfoFinder()
-            .getSnapshotInfo();
+    MigrationState state =
+        DatabaseMigrationStateSchedule.getValueAtTime(new SystemClock().nowUtc());
+    if (!state.getReplayDirection().equals(ReplayDirection.DATASTORE_TO_SQL)) {
+      throw new IllegalStateException("This pipeline is not designed for migration phase " + state);
+    }
 
-    new ValidateSqlPipeline(options, mostRecentExport).run(Pipeline.create(options));
+    LatestDatastoreSnapshotFinder datastoreSnapshotFinder =
+        DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
+            .datastoreSnapshotInfoFinder();
+
+    new ValidateSqlPipeline(options, datastoreSnapshotFinder).run(Pipeline.create(options));
   }
 }

core/src/main/java/google/registry/beam/comparedb/ValidateSqlPipelineOptions.java

@@ -16,7 +16,19 @@ package google.registry.beam.comparedb;
 
 import google.registry.beam.common.RegistryPipelineOptions;
 import google.registry.model.annotations.DeleteAfterMigration;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.options.Description;
 
 /** BEAM pipeline options for {@link ValidateSqlPipeline}. */
 @DeleteAfterMigration
-public interface ValidateSqlPipelineOptions extends RegistryPipelineOptions {}
+public interface ValidateSqlPipelineOptions extends RegistryPipelineOptions {
+
+  @Description(
+      "For history entries and EPP resources, only those modified strictly after this time are "
+          + "included in comparison. Value is in ISO8601 format. "
+          + "Other entity types are not affected.")
+  @Nullable
+  String getComparisonStartTimestamp();
+
+  void setComparisonStartTimestamp(String comparisonStartTimestamp);
+}

core/src/main/java/google/registry/beam/rde/RdeIO.java

@@ -248,6 +248,9 @@ public class RdeIO {
       // Now that we're done, output roll the cursor forward.
       if (key.manual()) {
         logger.atInfo().log("Manual operation; not advancing cursor or enqueuing upload task.");
+        // Temporary measure to run RDE in beam in parallel with the daily MapReduce based RDE runs.
+      } else if (tm().isOfy()) {
+        logger.atInfo().log("Ofy is primary TM; not advancing cursor or enqueuing upload task.");
       } else {
         outputReceiver.output(KV.of(key, revision));
       }

core/src/main/java/google/registry/config/RegistryConfig.java

@@ -33,6 +33,7 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSortedMap;
 import dagger.Module;
 import dagger.Provides;
+import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.util.TaskQueueUtils;
 import google.registry.util.YamlUtils;
 import java.lang.annotation.Documented;
@@ -392,19 +393,26 @@ public final class RegistryConfig {
 
     @Provides
     @Config("cloudSqlJdbcUrl")
-    public static String providesCloudSqlJdbcUrl(RegistryConfigSettings config) {
+    public static String provideCloudSqlJdbcUrl(RegistryConfigSettings config) {
       return config.cloudSql.jdbcUrl;
     }
 
     @Provides
     @Config("cloudSqlInstanceConnectionName")
-    public static String providesCloudSqlInstanceConnectionName(RegistryConfigSettings config) {
+    public static String provideCloudSqlInstanceConnectionName(RegistryConfigSettings config) {
       return config.cloudSql.instanceConnectionName;
     }
 
+    @Provides
+    @Config("cloudSqlReplicaInstanceConnectionName")
+    public static Optional<String> provideCloudSqlReplicaInstanceConnectionName(
+        RegistryConfigSettings config) {
+      return Optional.ofNullable(config.cloudSql.replicaInstanceConnectionName);
+    }
+
     @Provides
     @Config("cloudSqlDbInstanceName")
-    public static String providesCloudSqlDbInstance(RegistryConfigSettings config) {
+    public static String provideCloudSqlDbInstance(RegistryConfigSettings config) {
       // Format of instanceConnectionName: project-id:region:instance-name
       int lastColonIndex = config.cloudSql.instanceConnectionName.lastIndexOf(':');
       return config.cloudSql.instanceConnectionName.substring(lastColonIndex + 1);
@@ -1524,6 +1532,31 @@ public final class RegistryConfig {
     return CONFIG_SETTINGS.get().hibernate.hikariIdleTimeout;
   }
 
+  /**
+   * JDBC-specific: driver default batch size is 0, which means that every INSERT statement will be
+   * sent to the database individually. Batching allows us to group together multiple inserts into
+   * one single INSERT statement which can dramatically increase speed in situations with many
+   * inserts.
+   *
+   * <p>Hibernate docs, i.e.
+   * https://docs.jboss.org/hibernate/orm/5.6/userguide/html_single/Hibernate_User_Guide.html,
+   * recommend between 10 and 50.
+   */
+  public static String getHibernateJdbcBatchSize() {
+    return CONFIG_SETTINGS.get().hibernate.jdbcBatchSize;
+  }
+
+  /**
+   * Returns the JDBC fetch size.
+   *
+   * <p>Postgresql-specific: driver default fetch size is 0, which disables streaming result sets.
+   * Here we set a small default geared toward Nomulus server transactions. Large queries can
+   * override the defaults using {@link JpaTransactionManager#setQueryFetchSize}.
+   */
+  public static String getHibernateJdbcFetchSize() {
+    return CONFIG_SETTINGS.get().hibernate.jdbcFetchSize;
+  }
+
   /** Returns the roid suffix to be used for the roids of all contacts and hosts. */
   public static String getContactAndHostRoidSuffix() {
     return CONFIG_SETTINGS.get().registryPolicy.contactAndHostRoidSuffix;

core/src/main/java/google/registry/config/RegistryConfigSettings.java

@@ -120,6 +120,8 @@ public class RegistryConfigSettings {
     public String hikariMinimumIdle;
     public String hikariMaximumPoolSize;
     public String hikariIdleTimeout;
+    public String jdbcBatchSize;
+    public String jdbcFetchSize;
   }
 
   /** Configuration for Cloud SQL. */
@@ -128,6 +130,7 @@ public class RegistryConfigSettings {
     // TODO(05012021): remove username field after it is removed from all yaml files.
     public String username;
     public String instanceConnectionName;
+    public String replicaInstanceConnectionName;
   }
 
   /** Configuration for Apache Beam (Cloud Dataflow). */

core/src/main/java/google/registry/config/files/default-config.yaml

@@ -221,6 +221,17 @@ hibernate:
   hikariMinimumIdle: 1
   hikariMaximumPoolSize: 10
   hikariIdleTimeout: 300000
+  # The batch size is basically the number of insertions / updates in a single
+  # transaction that will be batched together into one INSERT/UPDATE statement.
+  # A larger batch size is useful when inserting or updating many entities in a
+  # single transaction. Hibernate docs
+  # (https://docs.jboss.org/hibernate/orm/5.6/userguide/html_single/Hibernate_User_Guide.html)
+  # recommend between 10 and 50.
+  jdbcBatchSize: 50
+  # The fetch size is the number of entities retrieved at a time from the
+  # database cursor. Here we set a small default geared toward Nomulus server
+  # transactions. Large queries can override the defaults on a per-query basis.
+  jdbcFetchSize: 20
 
 cloudSql:
   # jdbc url for the Cloud SQL database.
@@ -231,6 +242,10 @@ cloudSql:
   jdbcUrl: jdbc:postgresql://localhost
   # This name is used by Cloud SQL when connecting to the database.
   instanceConnectionName: project-id:region:instance-id
+  # If non-null, we will use this instance for certain read-only actions or
+  # pipelines, e.g. RDE, in order to offload some work from the primary
+  # instance. Expect any write actions on this instance to fail.
+  replicaInstanceConnectionName: null
 
 cloudDns:
   # Set both properties to null in Production.

core/src/main/java/google/registry/env/production/default/WEB-INF/cron.xml

@@ -36,6 +36,19 @@
     <target>backend</target>
   </cron>
 
+  <cron>
+    <url>/_dr/task/rdeStaging?beam=true</url>
+    <description>
+      This job generates a full RDE escrow deposit as a single gigantic XML
+      document using the Beam pipeline regardless of the current TM
+      configuration and streams it to cloud storage. It does not trigger the
+      subsequent upload tasks and is meant to run parallel with the main cron
+      job in order to compare the results from both runs.
+    </description>
+    <schedule>every 8 hours from 00:07 to 20:00</schedule>
+    <target>backend</target>
+  </cron>
+
   <cron>
     <url><![CDATA[/_dr/cron/fanout?queue=rde-upload&endpoint=/_dr/task/rdeUpload&forEachRealTld]]></url>
     <description>
@@ -349,6 +362,15 @@
     <target>backend</target>
   </cron>
 
+  <cron>
+    <url><![CDATA[/_dr/cron/replicateToDatastore]]></url>
+    <description>
+      Replays recent transactions from SQL to the Datastore secondary backend.
+    </description>
+    <schedule>every 3 minutes</schedule>
+    <target>backend</target>
+  </cron>
+
   <cron>
     <url><![CDATA[/_dr/task/wipeOutContactHistoryPii]]></url>
     <description>

core/src/main/java/google/registry/export/ExportPremiumTermsAction.java

@@ -26,7 +26,6 @@ import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSortedSet;
 import com.google.common.collect.Iterables;
-import com.google.common.collect.Streams;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
 import google.registry.config.RegistryConfig.Config;
@@ -143,7 +142,7 @@ public class ExportPremiumTermsAction implements Runnable {
         PremiumListDao.getLatestRevision(premiumListName).isPresent(),
         "Could not load premium list for " + tld);
     SortedSet<String> premiumTerms =
-        Streams.stream(PremiumListDao.loadAllPremiumEntries(premiumListName))
+        PremiumListDao.loadAllPremiumEntries(premiumListName).stream()
             .map(PremiumEntry::toString)
             .collect(ImmutableSortedSet.toImmutableSortedSet(String::compareTo));
 

core/src/main/java/google/registry/flows/contact/ContactCheckFlow.java

@@ -56,9 +56,9 @@ public final class ContactCheckFlow implements Flow {
   @Inject ContactCheckFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ImmutableList<String> targetIds = ((Check) resourceCommand).getTargetIds();
     verifyTargetIdCount(targetIds, maxChecks);
     ImmutableSet<String> existingIds =

core/src/main/java/google/registry/flows/contact/ContactCreateFlow.java

@@ -70,10 +70,10 @@ public final class ContactCreateFlow implements TransactionalFlow {
   @Inject ContactCreateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Create command = (Create) resourceCommand;
     DateTime now = tm().getTransactionTime();
     verifyResourceDoesNotExist(ContactResource.class, targetId, now, registrarId);

core/src/main/java/google/registry/flows/contact/ContactDeleteFlow.java

@@ -90,10 +90,10 @@ public final class ContactDeleteFlow implements TransactionalFlow {
   ContactDeleteFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     checkLinkedDomains(targetId, now, ContactResource.class, DomainBase::getReferencedContacts);
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);

core/src/main/java/google/registry/flows/contact/ContactInfoFlow.java

@@ -65,10 +65,10 @@ public final class ContactInfoFlow implements Flow {
   ContactInfoFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     DateTime now = clock.nowUtc();
-    extensionManager.validate(); // There are no legal extensions for this flow.
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ContactResource contact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     if (!isSuperuser) {
       verifyResourceOwnership(registrarId, contact);

core/src/main/java/google/registry/flows/contact/ContactTransferApproveFlow.java

@@ -74,14 +74,14 @@ public final class ContactTransferApproveFlow implements TransactionalFlow {
   @Inject ContactTransferApproveFlow() {}
 
   /**
-   * <p>The logic in this flow, which handles client approvals, very closely parallels the logic in
+   * The logic in this flow, which handles client approvals, very closely parallels the logic in
    * {@link ContactResource#cloneProjectedAtTime} which handles implicit server approvals.
    */
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingContact);

core/src/main/java/google/registry/flows/contact/ContactTransferCancelFlow.java

@@ -76,8 +76,8 @@ public final class ContactTransferCancelFlow implements TransactionalFlow {
   @Override
   public final EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingContact);

core/src/main/java/google/registry/flows/contact/ContactTransferQueryFlow.java

@@ -63,9 +63,9 @@ public final class ContactTransferQueryFlow implements Flow {
   @Inject ContactTransferQueryFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ContactResource contact =
         loadAndVerifyExistence(ContactResource.class, targetId, clock.nowUtc());
     verifyOptionalAuthInfo(authInfo, contact);

core/src/main/java/google/registry/flows/contact/ContactTransferRejectFlow.java

@@ -72,10 +72,10 @@ public final class ContactTransferRejectFlow implements TransactionalFlow {
   @Inject ContactTransferRejectFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingContact);

core/src/main/java/google/registry/flows/contact/ContactTransferRequestFlow.java

@@ -92,10 +92,10 @@ public final class ContactTransferRequestFlow implements TransactionalFlow {
   @Inject ContactTransferRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(gainingClientId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyAuthInfoPresentForResourceTransfer(authInfo);

core/src/main/java/google/registry/flows/contact/ContactUpdateFlow.java

@@ -89,10 +89,10 @@ public final class ContactUpdateFlow implements TransactionalFlow {
   @Inject ContactUpdateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Update command = (Update) resourceCommand;
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);

core/src/main/java/google/registry/flows/domain/DomainCheckFlow.java

@@ -135,8 +135,8 @@ public final class DomainCheckFlow implements Flow {
     extensionManager.register(
         FeeCheckCommandExtension.class, LaunchCheckExtension.class, AllocationTokenExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     ImmutableList<String> domainNames = ((Check) resourceCommand).getTargetIds();
     verifyTargetIdCount(domainNames, maxChecks);
     DateTime now = clock.nowUtc();

core/src/main/java/google/registry/flows/domain/DomainClaimsCheckFlow.java

@@ -82,8 +82,8 @@ public final class DomainClaimsCheckFlow implements Flow {
   @Override
   public EppResponse run() throws EppException {
     extensionManager.register(LaunchCheckExtension.class, AllocationTokenExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     if (eppInput.getSingleExtension(AllocationTokenExtension.class).isPresent()) {
       throw new DomainClaimsCheckNotAllowedWithAllocationTokens();
     }

core/src/main/java/google/registry/flows/domain/DomainCreateFlow.java

@@ -198,7 +198,7 @@ import org.joda.time.Duration;
  * @error {@link DomainPricingLogic.AllocationTokenInvalidForPremiumNameException}
  */
 @ReportingSpec(ActivityReportField.DOMAIN_CREATE)
-public class DomainCreateFlow implements TransactionalFlow {
+public final class DomainCreateFlow implements TransactionalFlow {
 
   /** Anchor tenant creates should always be for 2 years, since they get 2 years free. */
   private static final int ANCHOR_TENANT_CREATE_VALID_YEARS = 2;
@@ -219,7 +219,7 @@ public class DomainCreateFlow implements TransactionalFlow {
   @Inject DomainCreateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         FeeCreateCommandExtension.class,
         SecDnsCreateExtension.class,
@@ -227,9 +227,9 @@ public class DomainCreateFlow implements TransactionalFlow {
         LaunchCreateExtension.class,
         AllocationTokenExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
     verifyRegistrarIsActive(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainCommand.Create command = cloneAndLinkReferences((Create) resourceCommand, now);
     Period period = command.getPeriod();

core/src/main/java/google/registry/flows/domain/DomainDeleteFlow.java

@@ -138,12 +138,12 @@ public final class DomainDeleteFlow implements TransactionalFlow {
   @Inject DomainDeleteFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         MetadataExtension.class, SecDnsCreateExtension.class, DomainDeleteSuperuserExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     // Loads the target resource if it exists
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);

core/src/main/java/google/registry/flows/domain/DomainInfoFlow.java

@@ -91,11 +91,11 @@ public final class DomainInfoFlow implements Flow {
   DomainInfoFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(FeeInfoCommandExtensionV06.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = clock.nowUtc();
     DomainBase domain = verifyExistence(
         DomainBase.class, targetId, loadByForeignKey(DomainBase.class, targetId, now));

core/src/main/java/google/registry/flows/domain/DomainRenewFlow.java

@@ -135,12 +135,12 @@ public final class DomainRenewFlow implements TransactionalFlow {
   @Inject DomainRenewFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(FeeRenewCommandExtension.class, MetadataExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
     verifyRegistrarIsActive(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     Renew command = (Renew) resourceCommand;
     // Loads the target resource if it exists

core/src/main/java/google/registry/flows/domain/DomainRestoreRequestFlow.java

@@ -128,14 +128,14 @@ public final class DomainRestoreRequestFlow implements TransactionalFlow {
   @Inject DomainRestoreRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         FeeUpdateCommandExtension.class,
         MetadataExtension.class,
         RgpUpdateExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
     verifyRegistrarIsActive(registrarId);
+    extensionManager.validate();
     Update command = (Update) resourceCommand;
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);

core/src/main/java/google/registry/flows/domain/DomainTransferApproveFlow.java

@@ -99,14 +99,14 @@ public final class DomainTransferApproveFlow implements TransactionalFlow {
   @Inject DomainTransferApproveFlow() {}
 
   /**
-   * <p>The logic in this flow, which handles client approvals, very closely parallels the logic in
+   * The logic in this flow, which handles client approvals, very closely parallels the logic in
    * {@link DomainBase#cloneProjectedAtTime} which handles implicit server approvals.
    */
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingDomain);

core/src/main/java/google/registry/flows/domain/DomainTransferCancelFlow.java

@@ -86,10 +86,10 @@ public final class DomainTransferCancelFlow implements TransactionalFlow {
   @Inject DomainTransferCancelFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingDomain);

core/src/main/java/google/registry/flows/domain/DomainTransferQueryFlow.java

@@ -67,9 +67,9 @@ public final class DomainTransferQueryFlow implements Flow {
   @Inject DomainTransferQueryFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     DateTime now = clock.nowUtc();
     DomainBase domain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, domain);

core/src/main/java/google/registry/flows/domain/DomainTransferRejectFlow.java

@@ -88,10 +88,10 @@ public final class DomainTransferRejectFlow implements TransactionalFlow {
   @Inject DomainTransferRejectFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     Registry registry = Registry.get(existingDomain.getTld());

core/src/main/java/google/registry/flows/domain/DomainTransferRequestFlow.java

@@ -139,14 +139,14 @@ public final class DomainTransferRequestFlow implements TransactionalFlow {
   @Inject DomainTransferRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         DomainTransferRequestSuperuserExtension.class,
         FeeTransferCommandExtension.class,
         MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(gainingClientId);
     verifyRegistrarIsActive(gainingClientId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     Optional<DomainTransferRequestSuperuserExtension> superuserExtension =

core/src/main/java/google/registry/flows/domain/DomainUpdateFlow.java

@@ -168,8 +168,8 @@ public final class DomainUpdateFlow implements TransactionalFlow {
         SecDnsUpdateExtension.class,
         DomainUpdateSuperuserExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     Update command = cloneAndLinkReferences((Update) resourceCommand, now);
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);

core/src/main/java/google/registry/flows/host/HostCheckFlow.java

@@ -56,9 +56,9 @@ public final class HostCheckFlow implements Flow {
   @Inject HostCheckFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ImmutableList<String> hostnames = ((Check) resourceCommand).getTargetIds();
     verifyTargetIdCount(hostnames, maxChecks);
     ImmutableSet<String> existingIds =

core/src/main/java/google/registry/flows/host/HostCreateFlow.java

@@ -100,10 +100,10 @@ public final class HostCreateFlow implements TransactionalFlow {
   HostCreateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Create command = (Create) resourceCommand;
     DateTime now = tm().getTransactionTime();
     verifyResourceDoesNotExist(HostResource.class, targetId, now, registrarId);

core/src/main/java/google/registry/flows/host/HostDeleteFlow.java

@@ -92,10 +92,10 @@ public final class HostDeleteFlow implements TransactionalFlow {
   HostDeleteFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     validateHostName(targetId);
     checkLinkedDomains(targetId, now, HostResource.class, DomainBase::getNameservers);

core/src/main/java/google/registry/flows/host/HostInfoFlow.java

@@ -62,8 +62,8 @@ public final class HostInfoFlow implements Flow {
 
   @Override
   public EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     validateHostName(targetId);
     DateTime now = clock.nowUtc();
     HostResource host = loadAndVerifyExistence(HostResource.class, targetId, now);

core/src/main/java/google/registry/flows/host/HostUpdateFlow.java

@@ -125,10 +125,10 @@ public final class HostUpdateFlow implements TransactionalFlow {
   @Inject HostUpdateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Update command = (Update) resourceCommand;
     Change change = command.getInnerChange();
     String suppliedNewHostName = change.getFullyQualifiedHostName();

core/src/main/java/google/registry/flows/poll/PollAckFlow.java

@@ -45,17 +45,17 @@ import org.joda.time.DateTime;
 /**
  * An EPP flow for acknowledging {@link PollMessage}s.
  *
- * <p>Registrars refer to poll messages using an externally visible id generated by
- * {@link PollMessageExternalKeyConverter}. One-time poll messages are deleted from Datastore once
- * they are ACKed, whereas autorenew poll messages are simply marked as read, and won't be delivered
- * again until the next year of their recurrence.
+ * <p>Registrars refer to poll messages using an externally visible id generated by {@link
+ * PollMessageExternalKeyConverter}. One-time poll messages are deleted from Datastore once they are
+ * ACKed, whereas autorenew poll messages are simply marked as read, and won't be delivered again
+ * until the next year of their recurrence.
  *
  * @error {@link PollAckFlow.InvalidMessageIdException}
  * @error {@link PollAckFlow.MessageDoesNotExistException}
  * @error {@link PollAckFlow.MissingMessageIdException}
  * @error {@link PollAckFlow.NotAuthorizedToAckMessageException}
  */
-public class PollAckFlow implements TransactionalFlow {
+public final class PollAckFlow implements TransactionalFlow {
 
   @Inject ExtensionManager extensionManager;
   @Inject @RegistrarId String registrarId;
@@ -65,9 +65,9 @@ public class PollAckFlow implements TransactionalFlow {
   @Inject PollAckFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     if (messageId.isEmpty()) {
       throw new MissingMessageIdException();
     }

core/src/main/java/google/registry/flows/poll/PollRequestFlow.java

@@ -47,7 +47,7 @@ import org.joda.time.DateTime;
  *
  * @error {@link PollRequestFlow.UnexpectedMessageIdException}
  */
-public class PollRequestFlow implements Flow {
+public final class PollRequestFlow implements Flow {
 
   @Inject ExtensionManager extensionManager;
   @Inject @RegistrarId String registrarId;
@@ -57,9 +57,9 @@ public class PollRequestFlow implements Flow {
   @Inject PollRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     if (!messageId.isEmpty()) {
       throw new UnexpectedMessageIdException();
     }

core/src/main/java/google/registry/flows/session/HelloFlow.java

@@ -27,7 +27,7 @@ import javax.inject.Inject;
  *
  * @error {@link google.registry.flows.FlowUtils.GenericXmlSyntaxErrorException}
  */
-public class HelloFlow implements Flow {
+public final class HelloFlow implements Flow {
 
   @Inject ExtensionManager extensionManager;
   @Inject Clock clock;

core/src/main/java/google/registry/flows/session/LoginFlow.java

@@ -90,7 +90,7 @@ public class LoginFlow implements Flow {
   }
 
   /** Run the flow without bothering to log errors. The {@link #run} method will do that for us. */
-  private final EppResponse runWithoutLogging() throws EppException {
+  private EppResponse runWithoutLogging() throws EppException {
     extensionManager.validate();  // There are no legal extensions for this flow.
     Login login = (Login) eppInput.getCommandWrapper().getCommand();
     if (!registrarId.isEmpty()) {

core/src/main/java/google/registry/flows/session/LogoutFlow.java

@@ -30,7 +30,7 @@ import javax.inject.Inject;
  *
  * @error {@link google.registry.flows.FlowUtils.NotLoggedInException}
  */
-public class LogoutFlow implements Flow {
+public final class LogoutFlow implements Flow {
 
   @Inject ExtensionManager extensionManager;
   @Inject @RegistrarId String registrarId;
@@ -39,9 +39,9 @@ public class LogoutFlow implements Flow {
   @Inject LogoutFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     sessionMetadata.invalidate();
     return responseBuilder.setResultFromCode(SUCCESS_AND_CLOSE).build();
   }

core/src/main/java/google/registry/model/EntityClasses.java

@@ -15,6 +15,7 @@
 package google.registry.model;
 
 import com.google.common.collect.ImmutableSet;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.common.Cursor;
 import google.registry.model.common.EntityGroupRoot;
@@ -45,6 +46,7 @@ import google.registry.model.server.ServerSecret;
 import google.registry.model.tld.Registry;
 
 /** Sets of classes of the Objectify-registered entities in use throughout the model. */
+@DeleteAfterMigration
 public final class EntityClasses {
 
   /** Set of entity classes. */

core/src/main/java/google/registry/model/EppResource.java

@@ -21,6 +21,7 @@ import static com.google.common.collect.Sets.union;
 import static google.registry.config.RegistryConfig.getEppResourceCachingDuration;
 import static google.registry.config.RegistryConfig.getEppResourceMaxCachedEntries;
 import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.nullToEmpty;
 import static google.registry.util.CollectionUtils.nullToEmptyImmutableCopy;
@@ -380,13 +381,13 @@ public abstract class EppResource extends BackupGroupRoot implements Buildable {
 
         @Override
         public EppResource load(VKey<? extends EppResource> key) {
-          return tm().doTransactionless(() -> tm().loadByKey(key));
+          return replicaTm().doTransactionless(() -> replicaTm().loadByKey(key));
         }
 
         @Override
         public Map<VKey<? extends EppResource>, EppResource> loadAll(
             Iterable<? extends VKey<? extends EppResource>> keys) {
-          return tm().doTransactionless(() -> tm().loadByKeys(keys));
+          return replicaTm().doTransactionless(() -> replicaTm().loadByKeys(keys));
         }
       };
 

core/src/main/java/google/registry/model/IdService.java

@@ -20,6 +20,7 @@ import com.google.appengine.api.datastore.DatastoreServiceFactory;
 import com.google.common.annotations.VisibleForTesting;
 import google.registry.beam.common.RegistryPipelineWorkerInitializer;
 import google.registry.config.RegistryEnvironment;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -28,6 +29,7 @@ import java.util.concurrent.atomic.AtomicLong;
  * <p>In non-test, non-beam environments the Id is generated by Datastore, otherwise it's from an
  * atomic long number that's incremented every time this method is called.
  */
+@DeleteAfterMigration
 public final class IdService {
 
   /**

core/src/main/java/google/registry/model/SchemaVersion.java

@@ -19,12 +19,14 @@ import static com.google.common.base.Predicates.subtypeOf;
 import static java.util.stream.Collectors.joining;
 
 import com.google.common.collect.Ordering;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.ArrayDeque;
 import java.util.Queue;
 import java.util.SortedSet;
 import java.util.TreeSet;
 
 /** Utility methods for getting the version of the model schema from the model code. */
+@DeleteAfterMigration
 public final class SchemaVersion {
 
   /**

core/src/main/java/google/registry/model/common/CrossTldSingleton.java

@@ -20,11 +20,13 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.InCrossTld;
 import javax.persistence.MappedSuperclass;
 import javax.persistence.Transient;
 
 /** A singleton entity in Datastore. */
+@DeleteAfterMigration
 @MappedSuperclass
 @InCrossTld
 public abstract class CrossTldSingleton extends ImmutableObject {

core/src/main/java/google/registry/model/common/DatabaseMigrationStateSchedule.java

@@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryEnvironment;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.common.TimedTransitionProperty.TimedTransition;
 import google.registry.model.replay.SqlOnlyEntity;
 import java.time.Duration;
@@ -39,6 +40,7 @@ import org.joda.time.DateTime;
  * <p>The entity is stored in SQL throughout the entire migration so as to have a single point of
  * access.
  */
+@DeleteAfterMigration
 @Entity
 public class DatabaseMigrationStateSchedule extends CrossTldSingleton implements SqlOnlyEntity {
 

core/src/main/java/google/registry/model/common/EntityGroupRoot.java

@@ -19,6 +19,7 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import google.registry.model.BackupGroupRoot;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.replay.DatastoreOnlyEntity;
 import javax.annotation.Nullable;
 
@@ -37,6 +38,7 @@ import javax.annotation.Nullable;
  * entity group for the single namespace where global data applicable for all TLDs lived.
  */
 @Entity
+@DeleteAfterMigration
 public class EntityGroupRoot extends BackupGroupRoot implements DatastoreOnlyEntity {
 
   @SuppressWarnings("unused")

core/src/main/java/google/registry/model/index/EppResourceIndex.java

@@ -24,12 +24,15 @@ import com.googlecode.objectify.annotation.Index;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.BackupGroupRoot;
 import google.registry.model.EppResource;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.ReportedOn;
 import google.registry.model.replay.DatastoreOnlyEntity;
+import google.registry.persistence.VKey;
 
 /** An index that allows for quick enumeration of all EppResource entities (e.g. via map reduce). */
 @ReportedOn
 @Entity
+@DeleteAfterMigration
 public class EppResourceIndex extends BackupGroupRoot implements DatastoreOnlyEntity {
 
   @Id String id;
@@ -64,8 +67,9 @@ public class EppResourceIndex extends BackupGroupRoot implements DatastoreOnlyEn
     EppResourceIndex instance = instantiate(EppResourceIndex.class);
     instance.reference = resourceKey;
     instance.kind = resourceKey.getKind();
-    // TODO(b/207368050): figure out if this value has ever been used other than test cases
-    instance.id = resourceKey.getString(); // creates a web-safe key string
+    // creates a web-safe key string, this value is never used
+    // TODO(b/211785379): remove unused id
+    instance.id = VKey.from(resourceKey).stringify();
     instance.bucket = bucket;
     return instance;
   }

core/src/main/java/google/registry/model/index/EppResourceIndexBucket.java

@@ -23,12 +23,14 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import google.registry.model.EppResource;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.VirtualEntity;
 import google.registry.model.replay.DatastoreOnlyEntity;
 
 /** A virtual entity to represent buckets to which EppResourceIndex objects are randomly added. */
 @Entity
 @VirtualEntity
+@DeleteAfterMigration
 public class EppResourceIndexBucket extends ImmutableObject implements DatastoreOnlyEntity {
 
   @SuppressWarnings("unused")

core/src/main/java/google/registry/model/index/ForeignKeyIndex.java

@@ -21,6 +21,7 @@ import static google.registry.config.RegistryConfig.getEppResourceCachingDuratio
 import static google.registry.config.RegistryConfig.getEppResourceMaxCachedEntries;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaJpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.entriesToImmutableMap;
 import static google.registry.util.TypeUtils.instantiate;
@@ -43,6 +44,7 @@ import com.googlecode.objectify.annotation.Index;
 import google.registry.config.RegistryConfig;
 import google.registry.model.BackupGroupRoot;
 import google.registry.model.EppResource;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.ReportedOn;
 import google.registry.model.contact.ContactResource;
 import google.registry.model.domain.DomainBase;
@@ -50,6 +52,7 @@ import google.registry.model.host.HostResource;
 import google.registry.model.replay.DatastoreOnlyEntity;
 import google.registry.persistence.VKey;
 import google.registry.persistence.transaction.CriteriaQueryBuilder;
+import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.util.NonFinalForTesting;
 import java.util.Collection;
 import java.util.Comparator;
@@ -65,6 +68,7 @@ import org.joda.time.Duration;
  * the foreign key string. The instance is never deleted, but it is updated if a newer entity
  * becomes the active entity.
  */
+@DeleteAfterMigration
 public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroupRoot {
 
   /** The {@link ForeignKeyIndex} type for {@link ContactResource} entities. */
@@ -196,7 +200,7 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
    */
   public static <E extends EppResource> ImmutableMap<String, ForeignKeyIndex<E>> load(
       Class<E> clazz, Collection<String> foreignKeys, final DateTime now) {
-    return loadIndexesFromStore(clazz, foreignKeys, true).entrySet().stream()
+    return loadIndexesFromStore(clazz, foreignKeys, true, false).entrySet().stream()
         .filter(e -> now.isBefore(e.getValue().getDeletionTime()))
         .collect(entriesToImmutableMap());
   }
@@ -215,7 +219,10 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
    */
   private static <E extends EppResource>
       ImmutableMap<String, ForeignKeyIndex<E>> loadIndexesFromStore(
-          Class<E> clazz, Collection<String> foreignKeys, boolean inTransaction) {
+          Class<E> clazz,
+          Collection<String> foreignKeys,
+          boolean inTransaction,
+          boolean useReplicaJpaTm) {
     if (tm().isOfy()) {
       Class<ForeignKeyIndex<E>> fkiClass = mapToFkiClass(clazz);
       return ImmutableMap.copyOf(
@@ -224,17 +231,18 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
               : tm().doTransactionless(() -> auditedOfy().load().type(fkiClass).ids(foreignKeys)));
     } else {
       String property = RESOURCE_CLASS_TO_FKI_PROPERTY.get(clazz);
+      JpaTransactionManager jpaTmToUse = useReplicaJpaTm ? replicaJpaTm() : jpaTm();
       ImmutableList<ForeignKeyIndex<E>> indexes =
-          tm().transact(
-                  () ->
-                      jpaTm()
-                          .criteriaQuery(
-                              CriteriaQueryBuilder.create(clazz)
-                                  .whereFieldIsIn(property, foreignKeys)
-                                  .build())
-                          .getResultStream()
-                          .map(e -> ForeignKeyIndex.create(e, e.getDeletionTime()))
-                          .collect(toImmutableList()));
+          jpaTmToUse.transact(
+              () ->
+                  jpaTmToUse
+                      .criteriaQuery(
+                          CriteriaQueryBuilder.create(clazz)
+                              .whereFieldIsIn(property, foreignKeys)
+                              .build())
+                      .getResultStream()
+                      .map(e -> ForeignKeyIndex.create(e, e.getDeletionTime()))
+                      .collect(toImmutableList()));
       // We need to find and return the entities with the maximum deletionTime for each foreign key.
       return Multimaps.index(indexes, ForeignKeyIndex::getForeignKey).asMap().entrySet().stream()
           .map(
@@ -258,7 +266,8 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
               loadIndexesFromStore(
                       RESOURCE_CLASS_TO_FKI_CLASS.inverse().get(key.getKind()),
                       ImmutableSet.of(foreignKey),
-                      false)
+                      false,
+                      true)
                   .get(foreignKey));
         }
 
@@ -274,7 +283,7 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
               Streams.stream(keys).map(v -> v.getSqlKey().toString()).collect(toImmutableSet());
           ImmutableSet<VKey<ForeignKeyIndex<?>>> typedKeys = ImmutableSet.copyOf(keys);
           ImmutableMap<String, ? extends ForeignKeyIndex<? extends EppResource>> existingFkis =
-              loadIndexesFromStore(resourceClass, foreignKeys, false);
+              loadIndexesFromStore(resourceClass, foreignKeys, false, true);
           // ofy omits keys that don't have values in Datastore, so re-add them in
           // here with Optional.empty() values.
           return Maps.asMap(
@@ -334,7 +343,7 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
     // Safe to cast VKey<FKI<E>> to VKey<FKI<?>>
     @SuppressWarnings("unchecked")
     ImmutableList<VKey<ForeignKeyIndex<?>>> fkiVKeys =
-        Streams.stream(foreignKeys)
+        foreignKeys.stream()
             .map(fk -> (VKey<ForeignKeyIndex<?>>) VKey.create(fkiClass, fk))
             .collect(toImmutableList());
     try {

core/src/main/java/google/registry/model/ofy/AugmentedDeleter.java

@@ -23,6 +23,7 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Result;
 import com.googlecode.objectify.cmd.DeleteType;
 import com.googlecode.objectify.cmd.Deleter;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Arrays;
 import java.util.stream.Stream;
 
@@ -30,6 +31,7 @@ import java.util.stream.Stream;
  * A Deleter that forwards to {@code auditedOfy().delete()}, but can be augmented via subclassing to
  * do custom processing on the keys to be deleted prior to their deletion.
  */
+@DeleteAfterMigration
 abstract class AugmentedDeleter implements Deleter {
   private final Deleter delegate = ofy().delete();
 

core/src/main/java/google/registry/model/ofy/AugmentedSaver.java

@@ -21,13 +21,15 @@ import com.google.common.collect.ImmutableList;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Result;
 import com.googlecode.objectify.cmd.Saver;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Arrays;
 import java.util.Map;
 
 /**
- * A Saver that forwards to {@code ofy().save()}, but can be augmented via subclassing to
- * do custom processing on the entities to be saved prior to their saving.
+ * A Saver that forwards to {@code ofy().save()}, but can be augmented via subclassing to do custom
+ * processing on the entities to be saved prior to their saving.
  */
+@DeleteAfterMigration
 abstract class AugmentedSaver implements Saver {
   private final Saver delegate = ofy().save();
 

core/src/main/java/google/registry/model/ofy/CommitLogBucket.java

@@ -31,6 +31,7 @@ import com.googlecode.objectify.annotation.Id;
 import google.registry.config.RegistryConfig;
 import google.registry.model.Buildable;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -51,6 +52,7 @@ import org.joda.time.DateTime;
  */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogBucket extends ImmutableObject implements Buildable, DatastoreOnlyEntity {
 
   /**

core/src/main/java/google/registry/model/ofy/CommitLogCheckpoint.java

@@ -25,6 +25,7 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -45,6 +46,7 @@ import org.joda.time.DateTime;
  */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogCheckpoint extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** Shared singleton parent entity for commit log checkpoints. */

core/src/main/java/google/registry/model/ofy/CommitLogCheckpointRoot.java

@@ -21,6 +21,7 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -29,6 +30,7 @@ import org.joda.time.DateTime;
 /** Singleton parent entity for all commit log checkpoints. */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogCheckpointRoot extends ImmutableObject implements DatastoreOnlyEntity {
 
   public static final long SINGLETON_ID = 1;  // There is always exactly one of these.

core/src/main/java/google/registry/model/ofy/CommitLogManifest.java

@@ -23,6 +23,7 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -39,6 +40,7 @@ import org.joda.time.DateTime;
  */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogManifest extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** Commit log manifests are parented on a random bucket. */

core/src/main/java/google/registry/model/ofy/CommitLogMutation.java

@@ -27,6 +27,7 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -34,6 +35,7 @@ import google.registry.model.replay.DatastoreOnlyEntity;
 /** Representation of a saved entity in a {@link CommitLogManifest} (not deletes). */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogMutation extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** The manifest this belongs to. */

core/src/main/java/google/registry/model/ofy/CommitLoggedWork.java

@@ -30,6 +30,7 @@ import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
 import google.registry.model.BackupGroupRoot;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.util.Clock;
 import java.util.HashSet;
 import java.util.Map;
@@ -39,6 +40,7 @@ import java.util.function.Supplier;
 import org.joda.time.DateTime;
 
 /** Wrapper for {@link Supplier} that associates a time with each attempt. */
+@DeleteAfterMigration
 class CommitLoggedWork<R> implements Runnable {
 
   private final Supplier<R> work;

core/src/main/java/google/registry/model/ofy/DatastoreTransactionManager.java

@@ -33,6 +33,7 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Result;
 import com.googlecode.objectify.cmd.Query;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.InCrossTld;
 import google.registry.model.contact.ContactHistory;
 import google.registry.model.domain.DomainHistory;
@@ -55,6 +56,7 @@ import javax.persistence.NonUniqueResultException;
 import org.joda.time.DateTime;
 
 /** Datastore implementation of {@link TransactionManager}. */
+@DeleteAfterMigration
 public class DatastoreTransactionManager implements TransactionManager {
 
   private Ofy injectedOfy;

core/src/main/java/google/registry/model/ofy/EntityWritePriorities.java

@@ -16,6 +16,7 @@ package google.registry.model.ofy;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
+import google.registry.model.annotations.DeleteAfterMigration;
 
 /**
  * Contains the mapping from class names to SQL-replay-write priorities.
@@ -26,6 +27,7 @@ import com.google.common.collect.ImmutableMap;
  * values represent an earlier write (and later delete). Higher-valued classes can have foreign keys
  * on lower-valued classes, but not vice versa.
  */
+@DeleteAfterMigration
 public class EntityWritePriorities {
 
   /**

core/src/main/java/google/registry/model/ofy/ObjectifyService.java

@@ -35,6 +35,7 @@ import google.registry.config.RegistryEnvironment;
 import google.registry.model.Buildable;
 import google.registry.model.EntityClasses;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.translators.BloomFilterOfStringTranslatorFactory;
 import google.registry.model.translators.CidrAddressBlockTranslatorFactory;
 import google.registry.model.translators.CommitLogRevisionsTranslatorFactory;
@@ -52,6 +53,7 @@ import google.registry.model.translators.VKeyTranslatorFactory;
  * objects. The class contains a static initializer to call factory().register(...) on all
  * persistable objects in this package.
  */
+@DeleteAfterMigration
 public class ObjectifyService {
 
   /** A singleton instance of our Ofy wrapper. */

core/src/main/java/google/registry/model/ofy/Ofy.java

@@ -37,6 +37,7 @@ import com.googlecode.objectify.ObjectifyFactory;
 import com.googlecode.objectify.cmd.Deleter;
 import com.googlecode.objectify.cmd.Loader;
 import com.googlecode.objectify.cmd.Saver;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.VirtualEntity;
 import google.registry.model.ofy.ReadOnlyWork.KillTransactionException;
@@ -59,6 +60,7 @@ import org.joda.time.Duration;
  * simpler to wrap {@link Objectify} rather than extend it because this way we can remove some
  * methods that we don't really want exposed and add some shortcuts.
  */
+@DeleteAfterMigration
 public class Ofy {
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();

core/src/main/java/google/registry/model/ofy/OfyFilter.java

@@ -14,6 +14,7 @@
 
 package google.registry.model.ofy;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.io.IOException;
 import javax.servlet.Filter;
 import javax.servlet.FilterChain;
@@ -23,6 +24,7 @@ import javax.servlet.ServletRequest;
 import javax.servlet.ServletResponse;
 
 /** A filter that statically registers types with Objectify. */
+@DeleteAfterMigration
 public class OfyFilter implements Filter {
 
   @Override

core/src/main/java/google/registry/model/ofy/ReadOnlyWork.java

@@ -14,10 +14,12 @@
 
 package google.registry.model.ofy;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.util.Clock;
 import java.util.function.Supplier;
 
 /** Wrapper for {@link Supplier} that disallows mutations and fails the transaction at the end. */
+@DeleteAfterMigration
 class ReadOnlyWork<R> extends CommitLoggedWork<R> {
 
   ReadOnlyWork(Supplier<R> work, Clock clock) {

core/src/main/java/google/registry/model/ofy/ReplayQueue.java

@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableMap;
 import com.googlecode.objectify.Key;
 import google.registry.config.RegistryEnvironment;
 import google.registry.model.UpdateAutoTimestamp;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.replay.DatastoreEntity;
 import google.registry.model.replay.ReplaySpecializer;
 import google.registry.persistence.VKey;
@@ -34,6 +35,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
  *
  * <p>This code is to be removed when the actual replay cron job is implemented.
  */
+@DeleteAfterMigration
 public class ReplayQueue {
 
   static ConcurrentLinkedQueue<ImmutableMap<Key<?>, Object>> queue =

core/src/main/java/google/registry/model/ofy/RequestCapturingAsyncDatastoreService.java

@@ -28,6 +28,7 @@ import com.google.appengine.api.datastore.Query;
 import com.google.appengine.api.datastore.Transaction;
 import com.google.appengine.api.datastore.TransactionOptions;
 import com.google.common.collect.ImmutableList;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -35,6 +36,7 @@ import java.util.Map;
 import java.util.concurrent.Future;
 
 /** A proxy for {@link AsyncDatastoreService} that exposes call counts. */
+@DeleteAfterMigration
 public class RequestCapturingAsyncDatastoreService implements AsyncDatastoreService {
 
   private final AsyncDatastoreService delegate;

core/src/main/java/google/registry/model/ofy/SessionKeyExposingObjectify.java

@@ -18,8 +18,10 @@ import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.ObjectifyFactory;
 import com.googlecode.objectify.impl.ObjectifyImpl;
+import google.registry.model.annotations.DeleteAfterMigration;
 
 /** Registry-specific Objectify subclass that exposes the keys used in the current session. */
+@DeleteAfterMigration
 public class SessionKeyExposingObjectify extends ObjectifyImpl<SessionKeyExposingObjectify> {
 
   public SessionKeyExposingObjectify(ObjectifyFactory factory) {

core/src/main/java/google/registry/model/ofy/TimestampInversionException.java

@@ -17,6 +17,7 @@ package google.registry.model.ofy;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Objectify;
 import google.registry.model.BackupGroupRoot;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Arrays;
 import java.util.Map;
 import org.joda.time.DateTime;
@@ -25,6 +26,7 @@ import org.joda.time.DateTime;
  * Exception when trying to write to Datastore with a timestamp that is inconsistent with a partial
  * ordering on transactions that touch the same entities.
  */
+@DeleteAfterMigration
 class TimestampInversionException extends RuntimeException {
 
   static String getFileAndLine(StackTraceElement callsite) {

core/src/main/java/google/registry/model/ofy/TransactionInfo.java

@@ -26,10 +26,12 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Map;
 import org.joda.time.DateTime;
 
 /** Metadata for an {@link Ofy} transaction that saves commit logs. */
+@DeleteAfterMigration
 public class TransactionInfo {
 
   @VisibleForTesting

core/src/main/java/google/registry/model/replay/DatastoreAndSqlEntity.java

@@ -14,9 +14,11 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /** An entity that has the same Java object representation in SQL and Datastore. */
+@DeleteAfterMigration
 public interface DatastoreAndSqlEntity extends DatastoreEntity, SqlEntity {
 
   @Override

core/src/main/java/google/registry/model/replay/DatastoreEntity.java

@@ -14,6 +14,7 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /**
@@ -24,6 +25,7 @@ import java.util.Optional;
  * transactions and data into the secondary SQL store during the first, Datastore-primary, phase of
  * the migration.
  */
+@DeleteAfterMigration
 public interface DatastoreEntity {
 
   Optional<SqlEntity> toSqlEntity();

core/src/main/java/google/registry/model/replay/DatastoreOnlyEntity.java

@@ -14,9 +14,11 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /** An entity that is only stored in Datastore, that should not be replayed to SQL. */
+@DeleteAfterMigration
 public interface DatastoreOnlyEntity extends DatastoreEntity {
   @Override
   default Optional<SqlEntity> toSqlEntity() {

core/src/main/java/google/registry/model/replay/LastSqlTransaction.java

@@ -22,9 +22,11 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 
 /** Datastore entity to keep track of the last SQL transaction imported into the datastore. */
 @Entity
+@DeleteAfterMigration
 public class LastSqlTransaction extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** The key for this singleton. */

core/src/main/java/google/registry/model/replay/NonReplicatedEntity.java

@@ -14,6 +14,7 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /**
@@ -21,6 +22,7 @@ import java.util.Optional;
  *
  * <p>We expect that this is a result of the entity being dually-written.
  */
+@DeleteAfterMigration
 public interface NonReplicatedEntity extends DatastoreEntity, SqlEntity {
 
   @Override

core/src/main/java/google/registry/model/replay/ReplaySpecializer.java

@@ -14,6 +14,7 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.persistence.VKey;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -25,6 +26,7 @@ import java.lang.reflect.Method;
  * not directly present in the other database. This class allows us to do that by using reflection
  * to invoke special class methods if they are present.
  */
+@DeleteAfterMigration
 public class ReplaySpecializer {
 
   public static void beforeSqlDelete(VKey<?> key) {

core/src/main/java/google/registry/model/replay/ReplicateToDatastoreAction.java

@@ -27,6 +27,7 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.flogger.FluentLogger;
 import google.registry.model.UpdateAutoTimestamp;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
@@ -53,6 +54,7 @@ import org.joda.time.Duration;
     automaticallyPrintOk = true,
     auth = Auth.AUTH_INTERNAL_OR_ADMIN)
 @VisibleForTesting
+@DeleteAfterMigration
 public class ReplicateToDatastoreAction implements Runnable {
   public static final String PATH = "/_dr/cron/replicateToDatastore";
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();

core/src/main/java/google/registry/model/replay/SqlEntity.java

@@ -16,6 +16,7 @@ package google.registry.model.replay;
 
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /**
@@ -25,6 +26,7 @@ import java.util.Optional;
  * <p>This will be used when replaying SQL transactions into Datastore, during the second,
  * SQL-primary, phase of the migration from Datastore to SQL.
  */
+@DeleteAfterMigration
 public interface SqlEntity {
 
   Optional<DatastoreEntity> toDatastoreEntity();

core/src/main/java/google/registry/model/replay/SqlOnlyEntity.java

@@ -14,9 +14,11 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /** An entity that is only stored in SQL, that should not be replayed to Datastore. */
+@DeleteAfterMigration
 public interface SqlOnlyEntity extends SqlEntity {
   @Override
   default Optional<DatastoreEntity> toDatastoreEntity() {

core/src/main/java/google/registry/model/replay/SqlReplayCheckpoint.java

@@ -17,12 +17,14 @@ package google.registry.model.replay;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.common.CrossTldSingleton;
 import javax.persistence.Column;
 import javax.persistence.Entity;
 import org.joda.time.DateTime;
 
 @Entity
+@DeleteAfterMigration
 public class SqlReplayCheckpoint extends CrossTldSingleton implements SqlOnlyEntity {
 
   @Column(nullable = false)

core/src/main/java/google/registry/model/tld/label/PremiumList.java

@@ -21,7 +21,6 @@ import static com.google.common.hash.Funnels.stringFunnel;
 
 import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Streams;
 import com.google.common.hash.BloomFilter;
 import google.registry.model.Buildable;
 import google.registry.model.ImmutableObject;
@@ -86,9 +85,8 @@ public final class PremiumList extends BaseDomainLabelList<BigDecimal, PremiumEn
    */
   public synchronized ImmutableMap<String, BigDecimal> getLabelsToPrices() {
     if (labelsToPrices == null) {
-      Iterable<PremiumEntry> entries = PremiumListDao.loadAllPremiumEntries(name);
       labelsToPrices =
-          Streams.stream(entries)
+          PremiumListDao.loadAllPremiumEntries(name).stream()
               .collect(
                   toImmutableMap(
                       PremiumEntry::getDomainLabel,

core/src/main/java/google/registry/model/tld/label/PremiumListDao.java

@@ -14,11 +14,13 @@
 
 package google.registry.model.tld.label;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static google.registry.config.RegistryConfig.getDomainLabelListCacheDuration;
 import static google.registry.config.RegistryConfig.getSingletonCachePersistDuration;
 import static google.registry.config.RegistryConfig.getStaticPremiumListMaxCachedEntries;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.util.CollectionUtils.isNullOrEmpty;
 
 import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
@@ -26,8 +28,8 @@ import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.CacheLoader.InvalidCacheLoadException;
 import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Streams;
 import google.registry.model.tld.label.PremiumList.PremiumEntry;
 import google.registry.util.NonFinalForTesting;
 import java.math.BigDecimal;
@@ -54,8 +56,7 @@ public class PremiumListDao {
    * <p>This is cached for a shorter duration because we need to periodically reload this entity to
    * check if a new revision has been published, and if so, then use that.
    *
-   * <p>We also cache the absence of premium lists with a given name to avoid unnecessary pointless
-   * lookups. Note that this cache is only applicable to PremiumList objects stored in SQL.
+   * <p>We also cache the absence of premium lists with a given name to avoid pointless lookups.
    */
   @NonFinalForTesting
   static LoadingCache<String, Optional<PremiumList>> premiumListCache =
@@ -153,27 +154,25 @@ public class PremiumListDao {
   }
 
   public static PremiumList save(String name, CurrencyUnit currencyUnit, List<String> inputData) {
+    checkArgument(!inputData.isEmpty(), "New premium list data cannot be empty");
     return save(PremiumListUtils.parseToPremiumList(name, currencyUnit, inputData));
   }
 
+  /** Saves the given premium list (and its premium list entries) to Cloud SQL. */
   public static PremiumList save(PremiumList premiumList) {
-    jpaTm().transact(() -> jpaTm().insert(premiumList));
-    premiumListCache.invalidate(premiumList.getName());
     jpaTm()
         .transact(
             () -> {
-              if (premiumList.getLabelsToPrices() != null) {
-                Optional<PremiumList> savedPremiumList =
-                    PremiumListDao.getLatestRevision(premiumList.getName());
+              jpaTm().insert(premiumList);
+              jpaTm().getEntityManager().flush(); // This populates the revisionId.
+              long revisionId = premiumList.getRevisionId();
+
+              if (!isNullOrEmpty(premiumList.getLabelsToPrices())) {
                 ImmutableSet.Builder<PremiumEntry> entries = new ImmutableSet.Builder<>();
-                premiumList.getLabelsToPrices().entrySet().stream()
+                premiumList
+                    .getLabelsToPrices()
                     .forEach(
-                        entry ->
-                            entries.add(
-                                PremiumEntry.create(
-                                    savedPremiumList.get().getRevisionId(),
-                                    entry.getValue(),
-                                    entry.getKey())));
+                        (key, value) -> entries.add(PremiumEntry.create(revisionId, value, key)));
                 jpaTm().insertAll(entries.build());
               }
             });
@@ -216,7 +215,7 @@ public class PremiumListDao {
    *
    * <p>This is an expensive operation and should only be used when the entire list is required.
    */
-  public static Iterable<PremiumEntry> loadPremiumEntries(PremiumList premiumList) {
+  public static List<PremiumEntry> loadPremiumEntries(PremiumList premiumList) {
     return jpaTm()
         .transact(
             () ->
@@ -253,15 +252,14 @@ public class PremiumListDao {
    *
    * <p>This is an expensive operation and should only be used when the entire list is required.
    */
-  public static Iterable<PremiumEntry> loadAllPremiumEntries(String premiumListName) {
+  public static ImmutableList<PremiumEntry> loadAllPremiumEntries(String premiumListName) {
     PremiumList premiumList =
         getLatestRevision(premiumListName)
             .orElseThrow(
                 () ->
                     new IllegalArgumentException(
                         String.format("No premium list with name %s.", premiumListName)));
-    Iterable<PremiumEntry> entries = loadPremiumEntries(premiumList);
-    return Streams.stream(entries)
+    return loadPremiumEntries(premiumList).stream()
         .map(
             premiumEntry ->
                 new PremiumEntry.Builder()

core/src/main/java/google/registry/model/tld/label/PremiumListUtils.java

@@ -14,7 +14,6 @@
 
 package google.registry.model.tld.label;
 
-import static com.google.common.base.Preconditions.checkArgument;
 import static org.joda.time.DateTimeZone.UTC;
 
 import com.google.common.collect.ImmutableMap;
@@ -38,7 +37,6 @@ public class PremiumListUtils {
             .setCreationTimestamp(DateTime.now(UTC))
             .build();
     ImmutableMap<String, PremiumEntry> prices = partialPremiumList.parse(inputData);
-    checkArgument(inputData.size() > 0, "Input cannot be empty");
     Map<String, BigDecimal> priceAmounts = Maps.transformValues(prices, PremiumEntry::getValue);
     return partialPremiumList.asBuilder().setLabelsToPrices(priceAmounts).build();
   }

core/src/main/java/google/registry/model/translators/CommitLogRevisionsTranslatorFactory.java

@@ -23,6 +23,7 @@ import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.collect.Ordering;
 import com.googlecode.objectify.Key;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.ofy.CommitLogManifest;
 import google.registry.persistence.transaction.Transaction;
 import org.joda.time.DateTime;
@@ -31,11 +32,12 @@ import org.joda.time.DateTime;
  * Objectify translator for {@code ImmutableSortedMap<DateTime, Key<CommitLogManifest>>} fields.
  *
  * <p>This translator is responsible for doing three things:
+ *
  * <ol>
- * <li>Translating the data into two lists of {@code Date} and {@code Key} objects, in a manner
- *   similar to {@code @Mapify}.
- * <li>Inserting a key to the transaction's {@link CommitLogManifest} on save.
- * <li>Truncating the map to include only the last key per day for the last 30 days.
+ *   <li>Translating the data into two lists of {@code Date} and {@code Key} objects, in a manner
+ *       similar to {@code @Mapify}.
+ *   <li>Inserting a key to the transaction's {@link CommitLogManifest} on save.
+ *   <li>Truncating the map to include only the last key per day for the last 30 days.
  * </ol>
  *
  * <p>This allows you to have a field on your model object that tracks historical revisions of
@@ -46,6 +48,7 @@ import org.joda.time.DateTime;
  *
  * @see google.registry.model.EppResource
  */
+@DeleteAfterMigration
 public final class CommitLogRevisionsTranslatorFactory
     extends ImmutableSortedMapTranslatorFactory<DateTime, Key<CommitLogManifest>> {
 

core/src/main/java/google/registry/module/pubapi/PubApiComponent.java

@@ -30,6 +30,7 @@ import google.registry.keyring.api.KeyModule;
 import google.registry.keyring.kms.KmsModule;
 import google.registry.module.pubapi.PubApiRequestComponent.PubApiRequestComponentModule;
 import google.registry.monitoring.whitebox.StackdriverModule;
+import google.registry.persistence.PersistenceModule;
 import google.registry.privileges.secretmanager.SecretManagerModule;
 import google.registry.request.Modules.Jackson2Module;
 import google.registry.request.Modules.NetHttpTransportModule;
@@ -56,6 +57,7 @@ import javax.inject.Singleton;
       KeyringModule.class,
       KmsModule.class,
       NetHttpTransportModule.class,
+      PersistenceModule.class,
       PubApiRequestComponentModule.class,
       SecretManagerModule.class,
       ServerTridProviderModule.class,

core/src/main/java/google/registry/persistence/PersistenceComponent.java

@@ -19,6 +19,7 @@ import google.registry.config.CredentialModule;
 import google.registry.config.RegistryConfig.ConfigModule;
 import google.registry.keyring.kms.KmsModule;
 import google.registry.persistence.PersistenceModule.AppEngineJpaTm;
+import google.registry.persistence.PersistenceModule.ReadOnlyReplicaJpaTm;
 import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.privileges.secretmanager.SecretManagerModule;
 import google.registry.util.UtilsModule;
@@ -40,4 +41,7 @@ public interface PersistenceComponent {
 
   @AppEngineJpaTm
   JpaTransactionManager appEngineJpaTransactionManager();
+
+  @ReadOnlyReplicaJpaTm
+  JpaTransactionManager readOnlyReplicaJpaTransactionManager();
 }

core/src/main/java/google/registry/persistence/PersistenceModule.java

@@ -20,6 +20,8 @@ import static google.registry.config.RegistryConfig.getHibernateHikariConnection
 import static google.registry.config.RegistryConfig.getHibernateHikariIdleTimeout;
 import static google.registry.config.RegistryConfig.getHibernateHikariMaximumPoolSize;
 import static google.registry.config.RegistryConfig.getHibernateHikariMinimumIdle;
+import static google.registry.config.RegistryConfig.getHibernateJdbcBatchSize;
+import static google.registry.config.RegistryConfig.getHibernateJdbcFetchSize;
 import static google.registry.config.RegistryConfig.getHibernateLogSqlQueries;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 
@@ -76,15 +78,9 @@ public abstract class PersistenceModule {
   public static final String HIKARI_DS_CLOUD_SQL_INSTANCE =
       "hibernate.hikari.dataSource.cloudSqlInstance";
 
-  /**
-   * Postgresql-specific: driver default fetch size is 0, which disables streaming result sets. Here
-   * we set a small default geared toward Nomulus server transactions. Large queries can override
-   * the defaults using {@link JpaTransactionManager#setQueryFetchSize}.
-   */
+  public static final String JDBC_BATCH_SIZE = "hibernate.jdbc.batch_size";
   public static final String JDBC_FETCH_SIZE = "hibernate.jdbc.fetch_size";
 
-  private static final int DEFAULT_SERVER_FETCH_SIZE = 20;
-
   @VisibleForTesting
   @Provides
   @DefaultHibernateConfigs
@@ -111,7 +107,8 @@ public abstract class PersistenceModule {
     properties.put(HIKARI_MAXIMUM_POOL_SIZE, getHibernateHikariMaximumPoolSize());
     properties.put(HIKARI_IDLE_TIMEOUT, getHibernateHikariIdleTimeout());
     properties.put(Environment.DIALECT, NomulusPostgreSQLDialect.class.getName());
-    properties.put(JDBC_FETCH_SIZE, Integer.toString(DEFAULT_SERVER_FETCH_SIZE));
+    properties.put(JDBC_BATCH_SIZE, getHibernateJdbcBatchSize());
+    properties.put(JDBC_FETCH_SIZE, getHibernateJdbcFetchSize());
     return properties.build();
   }
 
@@ -122,8 +119,11 @@ public abstract class PersistenceModule {
       @Config("cloudSqlJdbcUrl") String jdbcUrl,
       @Config("cloudSqlInstanceConnectionName") String instanceConnectionName,
       @DefaultHibernateConfigs ImmutableMap<String, String> defaultConfigs) {
-    return createPartialSqlConfigs(
-        jdbcUrl, instanceConnectionName, defaultConfigs, Optional.empty());
+    HashMap<String, String> overrides = Maps.newHashMap(defaultConfigs);
+    overrides.put(Environment.URL, jdbcUrl);
+    overrides.put(HIKARI_DS_SOCKET_FACTORY, "com.google.cloud.sql.postgres.SocketFactory");
+    overrides.put(HIKARI_DS_CLOUD_SQL_INSTANCE, instanceConnectionName);
+    return ImmutableMap.copyOf(overrides);
   }
 
   /**
@@ -184,22 +184,6 @@ public abstract class PersistenceModule {
     return ImmutableMap.copyOf(overrides);
   }
 
-  @VisibleForTesting
-  static ImmutableMap<String, String> createPartialSqlConfigs(
-      String jdbcUrl,
-      String instanceConnectionName,
-      ImmutableMap<String, String> defaultConfigs,
-      Optional<Provider<TransactionIsolationLevel>> isolationOverride) {
-    HashMap<String, String> overrides = Maps.newHashMap(defaultConfigs);
-    overrides.put(Environment.URL, jdbcUrl);
-    overrides.put(HIKARI_DS_SOCKET_FACTORY, "com.google.cloud.sql.postgres.SocketFactory");
-    overrides.put(HIKARI_DS_CLOUD_SQL_INSTANCE, instanceConnectionName);
-    isolationOverride
-        .map(Provider::get)
-        .ifPresent(override -> overrides.put(Environment.ISOLATION, override.name()));
-    return ImmutableMap.copyOf(overrides);
-  }
-
   /**
    * Provides a {@link Supplier} of single-use JDBC {@link Connection connections} that can manage
    * the database DDL schema.
@@ -280,6 +264,40 @@ public abstract class PersistenceModule {
     return new JpaTransactionManagerImpl(create(overrides), clock);
   }
 
+  @Provides
+  @Singleton
+  @ReadOnlyReplicaJpaTm
+  static JpaTransactionManager provideReadOnlyReplicaJpaTm(
+      SqlCredentialStore credentialStore,
+      @PartialCloudSqlConfigs ImmutableMap<String, String> cloudSqlConfigs,
+      @Config("cloudSqlReplicaInstanceConnectionName")
+          Optional<String> replicaInstanceConnectionName,
+      Clock clock) {
+    HashMap<String, String> overrides = Maps.newHashMap(cloudSqlConfigs);
+    setSqlCredential(credentialStore, new RobotUser(RobotId.NOMULUS), overrides);
+    replicaInstanceConnectionName.ifPresent(
+        name -> overrides.put(HIKARI_DS_CLOUD_SQL_INSTANCE, name));
+    overrides.put(
+        Environment.ISOLATION, TransactionIsolationLevel.TRANSACTION_READ_COMMITTED.name());
+    return new JpaTransactionManagerImpl(create(overrides), clock);
+  }
+
+  @Provides
+  @Singleton
+  @BeamReadOnlyReplicaJpaTm
+  static JpaTransactionManager provideBeamReadOnlyReplicaJpaTm(
+      @BeamPipelineCloudSqlConfigs ImmutableMap<String, String> beamCloudSqlConfigs,
+      @Config("cloudSqlReplicaInstanceConnectionName")
+          Optional<String> replicaInstanceConnectionName,
+      Clock clock) {
+    HashMap<String, String> overrides = Maps.newHashMap(beamCloudSqlConfigs);
+    replicaInstanceConnectionName.ifPresent(
+        name -> overrides.put(HIKARI_DS_CLOUD_SQL_INSTANCE, name));
+    overrides.put(
+        Environment.ISOLATION, TransactionIsolationLevel.TRANSACTION_READ_COMMITTED.name());
+    return new JpaTransactionManagerImpl(create(overrides), clock);
+  }
+
   /** Constructs the {@link EntityManagerFactory} instance. */
   @VisibleForTesting
   static EntityManagerFactory create(
@@ -357,7 +375,12 @@ public abstract class PersistenceModule {
      * The {@link JpaTransactionManager} optimized for bulk loading multi-level JPA entities. Please
      * see {@link google.registry.model.bulkquery.BulkQueryEntities} for more information.
      */
-    BULK_QUERY
+    BULK_QUERY,
+    /**
+     * The {@link JpaTransactionManager} that uses the read-only Postgres replica if configured, or
+     * the standard DB if not.
+     */
+    READ_ONLY_REPLICA
   }
 
   /** Dagger qualifier for JDBC {@link Connection} with schema management privilege. */
@@ -383,6 +406,22 @@ public abstract class PersistenceModule {
   @Documented
   public @interface BeamBulkQueryJpaTm {}
 
+  /**
+   * Dagger qualifier for {@link JpaTransactionManager} used inside BEAM pipelines that uses the
+   * read-only Postgres replica if one is configured (otherwise it uses the standard DB).
+   */
+  @Qualifier
+  @Documented
+  public @interface BeamReadOnlyReplicaJpaTm {}
+
+  /**
+   * Dagger qualifier for {@link JpaTransactionManager} that uses the read-only Postgres replica if
+   * one is configured (otherwise it uses the standard DB).
+   */
+  @Qualifier
+  @Documented
+  public @interface ReadOnlyReplicaJpaTm {}
+
   /** Dagger qualifier for {@link JpaTransactionManager} used for Nomulus tool. */
   @Qualifier
   @Documented

core/src/main/java/google/registry/persistence/VKey.java

@@ -144,7 +144,7 @@ public class VKey<T> extends ImmutableObject implements Serializable {
    */
   public static <T> VKey<T> create(String keyString) {
     if (!keyString.startsWith(CLASS_TYPE + KV_SEPARATOR)) {
-      // to handle the existing ofy key string
+      // handle the existing ofy key string
       return fromWebsafeKey(keyString);
     } else {
       ImmutableMap<String, String> kvs =
@@ -307,6 +307,7 @@ public class VKey<T> extends ImmutableObject implements Serializable {
     if (maybeGetSqlKey().isPresent()) {
       key += DELIMITER + SQL_LOOKUP_KEY + KV_SEPARATOR + SerializeUtils.stringify(getSqlKey());
     }
+    // getString() method returns a Base64 encoded web safe of ofy key
     if (maybeGetOfyKey().isPresent()) {
       key += DELIMITER + OFY_LOOKUP_KEY + KV_SEPARATOR + getOfyKey().getString();
     }

core/src/main/java/google/registry/persistence/converter/DatabaseMigrationScheduleTransitionConverter.java

@@ -15,6 +15,7 @@
 package google.registry.persistence.converter;
 
 import com.google.common.collect.Maps;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationStateTransition;
@@ -23,6 +24,7 @@ import javax.persistence.Converter;
 import org.joda.time.DateTime;
 
 /** JPA converter for {@link DatabaseMigrationStateSchedule} transitions. */
+@DeleteAfterMigration
 @Converter(autoApply = true)
 public class DatabaseMigrationScheduleTransitionConverter
     extends TimedTransitionPropertyConverterBase<MigrationState, MigrationStateTransition> {