Remove the use of AppEngineEnvironment in Spec11Pipeline (#1365)

After #1348 it is no longer necessary to use AppEngineEnvironment in
Beam pipelines. In tests it is taken care of by the
DatastoreEntityExtension whereas on Dataflow the
RegistryPipelineWorkerInitializer does the same initialization for Ofy.

build.gradle

@@ -196,8 +196,31 @@ allprojects {
   }
 }
 
+rootProject.ext {
+  pyver = { exe ->
+    try {
+      ext.execInBash(
+          exe + " -c 'import sys; print(sys.hexversion)'", "/") as Integer
+    } catch (org.gradle.process.internal.ExecException e) {
+      return -1;
+    }
+  }
+}
+
 task runPresubmits(type: Exec) {
-  executable '/usr/bin/python3'
+
+  // Find a python version greater than 3.7.3 (this is somewhat arbitrary, we
+  // know we'd like at least 3.6, but 3.7.3 is the latest that ships with
+  // Debian so it seems like that should be available anywhere).
+  def MIN_PY_VER = 0x3070300
+  if (pyver('python') >= MIN_PY_VER) {
+    executable 'python'
+  } else if (pyver('/usr/bin/python3') >= MIN_PY_VER) {
+    executable '/usr/bin/python3'
+  } else {
+    throw new GradleException("No usable Python version found (build " +
+                              "requires at least python 3.7.3)");
+  }
   args('config/presubmits.py')
 }
 
@@ -457,8 +480,6 @@ task javaIncrementalFormatApply {
 task javadoc(type: Javadoc) {
   source javadocSource
   classpath = files(javadocClasspath)
-  // Exclude the misbehaving generated-by-Soy Java files
-  exclude "**/*SoyInfo.java"
   destinationDir = file("${buildDir}/docs/javadoc")
   options.encoding = "UTF-8"
   // In a lot of places we don't write @return so suppress warnings about that.
@@ -483,3 +504,15 @@ task coreDev {
 }
 
 javadocDependentTasks.each { tasks.javadoc.dependsOn(it) }
+
+// disable javadoc in subprojects, these will break because they don't have
+// the correct classpath (see above).
+gradle.taskGraph.whenReady { graph ->
+  graph.getAllTasks().each { task ->
+    def subprojectJavadoc = (task.path =~ /:.+:javadoc/)
+    if (subprojectJavadoc) {
+        println "Skipping ${task.path} for javadoc (only root javadoc works)"
+        task.enabled = false
+    }
+  }
+}

common/src/testing/java/google/registry/testing/SystemInfo.java

@@ -39,7 +39,7 @@ public final class SystemInfo {
                     pid.getOutputStream().close();
                     pid.waitFor();
                   } catch (IOException e) {
-                    logger.atWarning().withCause(e).log("%s command not available", cmd);
+                    logger.atWarning().withCause(e).log("%s command not available.", cmd);
                     return false;
                   }
                   return true;

config/nom_build.py

@@ -140,6 +140,8 @@ PROPERTIES = [
              'a BEAM pipeline to image. Setting this property to empty string '
              'will disable image generation.',
              '/usr/bin/dot'),
+    Property('pipeline',
+             'The name of the Beam pipeline being staged.')
 ]
 
 GRADLE_FLAGS = [

config/presubmits.py

@@ -17,9 +17,11 @@ These aren't built in to the static code analysis tools we use (e.g. Checkstyle,
 Error Prone) so we must write them manually.
 """
 
+import json
 import os
 from typing import List, Tuple
 import sys
+import textwrap
 import re
 
 # We should never analyze any generated files
@@ -28,6 +30,13 @@ UNIVERSALLY_SKIPPED_PATTERNS = {"/build/", "cloudbuild-caches", "/out/", ".git/"
 FORBIDDEN = 1
 REQUIRED = 2
 
+# The list of expected json packages and their licenses.
+# These should be one of the allowed licenses in:
+# config/dependency-license/allowed_licenses.json
+EXPECTED_JS_PACKAGES = [
+    'google-closure-library',   # Owned by Google, Apache 2.0
+]
+
 
 class PresubmitCheck:
 
@@ -206,15 +215,12 @@ PRESUBMITS = {
          "RegistryJpaIO.java",
          # TODO(b/179158393): Remove everything below, which should be done
          # using Criteria
-         "ForeignKeyIndex.java",
-         "HistoryEntryDao.java",
          "JpaTransactionManager.java",
          "JpaTransactionManagerImpl.java",
          # CriteriaQueryBuilder is a false positive
          "CriteriaQueryBuilder.java",
          "RdapDomainSearchAction.java",
          "RdapNameserverSearchAction.java",
-         "RdapSearchActionBase.java",
          "ReadOnlyCheckingEntityManager.java",
          "RegistryQuery",
          },
@@ -311,6 +317,26 @@ def verify_flyway_index():
   return not success
 
 
+def verify_javascript_deps():
+  """Verifies that we haven't introduced any new javascript dependencies."""
+  with open('package.json') as f:
+    package = json.load(f)
+
+  deps = list(package['dependencies'].keys())
+  if deps != EXPECTED_JS_PACKAGES:
+    print('Unexpected javascript dependencies.  Was expecting '
+          '%s, got %s.' % (EXPECTED_JS_PACKAGES, deps))
+    print(textwrap.dedent("""
+        * If the new dependencies are intentional, please verify that the
+        * license is one of the allowed licenses (see
+        * config/dependency-license/allowed_licenses.json) and add an entry
+        * for the package (with the license in a comment) to the
+        * EXPECTED_JS_PACKAGES variable in config/presubmits.py.
+        """))
+    return True
+  return False
+
+
 def get_files():
   for root, dirnames, filenames in os.walk("."):
     for filename in filenames:
@@ -318,6 +344,7 @@ def get_files():
 
 
 if __name__ == "__main__":
+  print('python version is %s' % sys.version)
   failed = False
   for file in get_files():
     error_messages = []
@@ -334,5 +361,8 @@ if __name__ == "__main__":
   # when we put it here it fails fast before all of the tests are run.
   failed |= verify_flyway_index()
 
+  # Make sure we haven't introduced any javascript dependencies.
+  failed |= verify_javascript_deps()
+
   if failed:
     sys.exit(1)

core/build.gradle

@@ -457,6 +457,22 @@ task soyToJava {
               "--srcs", "${soyFiles.join(',')}",
               "--compileTimeGlobalsFile", "${resourcesSourceDir}/google/registry/ui/globals.txt"
     }
+
+    // Replace the "@link" tags after the "@deprecated" tags in the generated
+    // files.  The soy compiler doesn't generate imports for these, causing
+    // us to get warnings when we generate javadocs.
+    // TODO(b/200296387): To be fair, the deprecations are accurate: we're
+    // using the old "SoyInfo" classes instead of the new "Templates" files.
+    // When we convert to the new classes, this hack can go away.
+    def outputs = fileTree(outputDirectory) {
+      include '**/*.java'
+    }
+
+    outputs.each { file ->
+      exec {
+        commandLine 'sed', '-i', 's/@link/LINK/g', file.getCanonicalPath()
+      }
+    }
   }
 
   doLast {
@@ -756,50 +772,57 @@ createUberJar('nomulus', 'nomulus', 'google.registry.tools.RegistryTool')
 // This packages more code and dependency than necessary. However, without
 // restructuring the source tree it is difficult to generate leaner jars.
 createUberJar(
-    'beam_pipeline_common',
+    'beamPipelineCommon',
     'beam_pipeline_common',
     '')
 
-// Create beam staging task if environment is alpha or crash.
-// All other environments use formally released pipelines through CloudBuild.
+// Create beam staging task if the environment is alpha. Production, sandbox and
+// qa use formally released pipelines through CloudBuild, whereas crash and
+// alpha use the pipelines staged on alpha deployment project.
 //
 // User should install gcloud and login to GCP before invoking this tasks.
-if (environment in ['alpha', 'crash']) {
+if (environment == 'alpha') {
   def pipelines = [
-      [
-          mainClass: 'google.registry.beam.initsql.InitSqlPipeline',
-          metaData: 'google/registry/beam/init_sql_pipeline_metadata.json'
-      ],
-      [
-          mainClass: 'google.registry.beam.datastore.BulkDeleteDatastorePipeline',
-          metaData: 'google/registry/beam/bulk_delete_datastore_pipeline_metadata.json'
-      ],
-      [
-          mainClass: 'google.registry.beam.spec11.Spec11Pipeline',
-          metaData: 'google/registry/beam/spec11_pipeline_metadata.json'
-      ],
-      [
-          mainClass: 'google.registry.beam.invoicing.InvoicingPipeline',
-          metaData: 'google/registry/beam/invoicing_pipeline_metadata.json'
-      ],
-      [
-          mainClass: 'google.registry.beam.rde.RdePipeline',
-          metaData: 'google/registry/beam/rde_pipeline_metadata.json'
-      ],
+      InitSql            :
+          [
+              mainClass: 'google.registry.beam.initsql.InitSqlPipeline',
+              metaData : 'google/registry/beam/init_sql_pipeline_metadata.json'
+          ],
+      BulkDeleteDatastore:
+          [
+              mainClass: 'google.registry.beam.datastore.BulkDeleteDatastorePipeline',
+              metaData : 'google/registry/beam/bulk_delete_datastore_pipeline_metadata.json'
+          ],
+      Spec11             :
+          [
+              mainClass: 'google.registry.beam.spec11.Spec11Pipeline',
+              metaData : 'google/registry/beam/spec11_pipeline_metadata.json'
+          ],
+      Invoicing          :
+          [
+              mainClass: 'google.registry.beam.invoicing.InvoicingPipeline',
+              metaData : 'google/registry/beam/invoicing_pipeline_metadata.json'
+          ],
+      Rde                :
+          [
+              mainClass: 'google.registry.beam.rde.RdePipeline',
+              metaData : 'google/registry/beam/rde_pipeline_metadata.json'
+          ],
   ]
-  project.tasks.create("stage_beam_pipelines") {
+  project.tasks.create("stageBeamPipelines") {
     doLast {
       pipelines.each {
-        def mainClass = it['mainClass']
-        def metaData = it['metaData']
-        def pipelineName = CaseFormat.UPPER_CAMEL.to(
-                CaseFormat.LOWER_UNDERSCORE,
-                mainClass.substring(mainClass.lastIndexOf('.') + 1))
-        def imageName = "gcr.io/${gcpProject}/beam/${pipelineName}"
-        def metaDataBaseName = metaData.substring(metaData.lastIndexOf('/') + 1)
-        def uberJarName = tasks.beam_pipeline_common.outputs.files.asPath
+        if (rootProject.pipeline == ''|| rootProject.pipeline == it.key) {
+          def mainClass = it.value['mainClass']
+          def metaData = it.value['metaData']
+          def pipelineName = CaseFormat.UPPER_CAMEL.to(
+              CaseFormat.LOWER_UNDERSCORE,
+              mainClass.substring(mainClass.lastIndexOf('.') + 1))
+          def imageName = "gcr.io/${gcpProject}/beam/${pipelineName}"
+          def metaDataBaseName = metaData.substring(metaData.lastIndexOf('/') + 1)
+          def uberJarName = tasks.beamPipelineCommon.outputs.files.asPath
 
-        def command = "\
+          def command = "\
             gcloud dataflow flex-template build \
             gs://${gcpProject}-deploy/live/beam/${metaDataBaseName} \
             --image-gcr-path ${imageName}:live \
@@ -809,10 +832,11 @@ if (environment in ['alpha', 'crash']) {
             --jar ${uberJarName} \
             --env FLEX_TEMPLATE_JAVA_MAIN_CLASS=${mainClass} \
             --project ${gcpProject}".toString()
-        rootProject.ext.execInBash(command, '/tmp')
+          rootProject.ext.execInBash(command, '/tmp')
+        }
       }
     }
-  }.dependsOn(tasks.beam_pipeline_common)
+  }.dependsOn(tasks.beamPipelineCommon)
 }
 
 // A jar with classes and resources from main sourceSet, excluding internal
@@ -1087,6 +1111,10 @@ test {
   // TODO(weiminyu): Remove dependency on sqlIntegrationTest
 }.dependsOn(fragileTest, outcastTest, standardTest, registryToolIntegrationTest, sqlIntegrationTest)
 
+// When we override tests, we also break the cleanTest command.
+cleanTest.dependsOn(cleanFragileTest, cleanOutcastTest, cleanStandardTest,
+                    cleanRegistryToolIntegrationTest, cleanSqlIntegrationTest)
+
 project.build.dependsOn devtool
 project.build.dependsOn buildToolImage
 project.build.dependsOn ':stage'

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

@@ -57,8 +57,7 @@ public final class CommitLogImports {
    */
   public static ImmutableList<ImmutableList<VersionedEntity>> loadEntitiesByTransaction(
       InputStream inputStream) {
-    try (AppEngineEnvironment appEngineEnvironment = new AppEngineEnvironment();
-        InputStream input = new BufferedInputStream(inputStream)) {
+    try (InputStream input = new BufferedInputStream(inputStream)) {
       Iterator<ImmutableObject> commitLogs = createDeserializingIterator(input, false);
       checkState(commitLogs.hasNext());
       checkState(commitLogs.next() instanceof CommitLogCheckpoint);

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

@@ -93,7 +93,7 @@ public final class DeleteOldCommitLogsAction implements Runnable {
   public void run() {
     DateTime deletionThreshold = clock.nowUtc().minus(maxAge);
     logger.atInfo().log(
-        "Processing asynchronous deletion of unreferenced CommitLogManifests older than %s",
+        "Processing asynchronous deletion of unreferenced CommitLogManifests older than %s.",
         deletionThreshold);
 
     mrRunner
@@ -208,7 +208,7 @@ public final class DeleteOldCommitLogsAction implements Runnable {
       getContext().incrementCounter("EPP resources missing pre-threshold revision (SEE LOGS)");
       logger.atSevere().log(
           "EPP resource missing old enough revision: "
-              + "%s (created on %s) has %d revisions between %s and %s, while threshold is %s",
+              + "%s (created on %s) has %d revisions between %s and %s, while threshold is %s.",
           Key.create(eppResource),
           eppResource.getCreationTime(),
           eppResource.getRevisions().size(),

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

@@ -100,7 +100,7 @@ public final class ExportCommitLogDiffAction implements Runnable {
 
     // Load the keys of all the manifests to include in this diff.
     List<Key<CommitLogManifest>> sortedKeys = loadAllDiffKeys(lowerCheckpoint, upperCheckpoint);
-    logger.atInfo().log("Found %d manifests to export", sortedKeys.size());
+    logger.atInfo().log("Found %d manifests to export.", sortedKeys.size());
     // Open an output channel to GCS, wrapped in a stream for convenience.
     try (OutputStream gcsStream =
         gcsUtils.openOutputStream(
@@ -124,7 +124,7 @@ public final class ExportCommitLogDiffAction implements Runnable {
       for (int i = 0; i < keyChunks.size(); i++) {
         // Force the async load to finish.
         Collection<CommitLogManifest> chunkValues = nextChunkToExport.values();
-        logger.atInfo().log("Loaded %d manifests", chunkValues.size());
+        logger.atInfo().log("Loaded %d manifests.", chunkValues.size());
         // Since there is no hard bound on how much data this might be, take care not to let the
         // Objectify session cache fill up and potentially run out of memory. This is the only safe
         // point to do this since at this point there is no async load in progress.
@@ -134,12 +134,12 @@ public final class ExportCommitLogDiffAction implements Runnable {
           nextChunkToExport = auditedOfy().load().keys(keyChunks.get(i + 1));
         }
         exportChunk(gcsStream, chunkValues);
-        logger.atInfo().log("Exported %d manifests", chunkValues.size());
+        logger.atInfo().log("Exported %d manifests.", chunkValues.size());
       }
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
-    logger.atInfo().log("Exported %d manifests in total", sortedKeys.size());
+    logger.atInfo().log("Exported %d total manifests.", sortedKeys.size());
   }
 
   /**

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

@@ -94,14 +94,14 @@ class GcsDiffFileLister {
           logger.atInfo().log(
               "Gap discovered in sequence terminating at %s, missing file: %s",
               sequence.lastKey(), filename);
-          logger.atInfo().log("Found sequence from %s to %s", checkpointTime, lastTime);
+          logger.atInfo().log("Found sequence from %s to %s.", checkpointTime, lastTime);
           return false;
         }
       }
       sequence.put(checkpointTime, blobInfo);
       checkpointTime = getLowerBoundTime(blobInfo);
     }
-    logger.atInfo().log("Found sequence from %s to %s", checkpointTime, lastTime);
+    logger.atInfo().log("Found sequence from %s to %s.", checkpointTime, lastTime);
     return true;
   }
 
@@ -140,7 +140,7 @@ class GcsDiffFileLister {
         }
       }
       if (upperBoundTimesToBlobInfo.isEmpty()) {
-        logger.atInfo().log("No files found");
+        logger.atInfo().log("No files found.");
         return ImmutableList.of();
       }
 
@@ -185,7 +185,7 @@ class GcsDiffFileLister {
 
     logger.atInfo().log(
         "Actual restore from time: %s", getLowerBoundTime(sequence.firstEntry().getValue()));
-    logger.atInfo().log("Found %d files to restore", sequence.size());
+    logger.atInfo().log("Found %d files to restore.", sequence.size());
     return ImmutableList.copyOf(sequence.values());
   }
 

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

@@ -273,7 +273,7 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
             ofyPojo.getClass());
       }
     } catch (Throwable t) {
-      logger.atSevere().log("Error when replaying object %s", ofyPojo);
+      logger.atSevere().log("Error when replaying object %s.", ofyPojo);
       throw t;
     }
   }
@@ -300,7 +300,7 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
         jpaTm().deleteIgnoringReadOnly(entityVKey);
       }
     } catch (Throwable t) {
-      logger.atSevere().log("Error when deleting key %s", entityVKey);
+      logger.atSevere().log("Error when deleting key %s.", entityVKey);
       throw t;
     }
   }

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

@@ -103,13 +103,13 @@ public class RestoreCommitLogsAction implements Runnable {
         !FORBIDDEN_ENVIRONMENTS.contains(RegistryEnvironment.get()),
         "DO NOT RUN IN PRODUCTION OR SANDBOX.");
     if (dryRun) {
-      logger.atInfo().log("Running in dryRun mode");
+      logger.atInfo().log("Running in dry-run mode.");
     }
     String gcsBucket = gcsBucketOverride.orElse(defaultGcsBucket);
     logger.atInfo().log("Restoring from %s.", gcsBucket);
     List<BlobInfo> diffFiles = diffLister.listDiffFiles(gcsBucket, fromTime, toTime);
     if (diffFiles.isEmpty()) {
-      logger.atInfo().log("Nothing to restore");
+      logger.atInfo().log("Nothing to restore.");
       return;
     }
     Map<Integer, DateTime> bucketTimestamps = new HashMap<>();
@@ -143,7 +143,7 @@ public class RestoreCommitLogsAction implements Runnable {
                                 .build()),
                 Stream.of(CommitLogCheckpointRoot.create(lastCheckpoint.getCheckpointTime())))
             .collect(toImmutableList()));
-    logger.atInfo().log("Restore complete");
+    logger.atInfo().log("Restore complete.");
   }
 
   /**

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

@@ -24,6 +24,7 @@ import static google.registry.batch.AsyncTaskEnqueuer.QUEUE_ASYNC_HOST_RENAME;
 import static google.registry.request.RequestParameters.extractIntParameter;
 import static google.registry.request.RequestParameters.extractLongParameter;
 import static google.registry.request.RequestParameters.extractOptionalBooleanParameter;
+import static google.registry.request.RequestParameters.extractOptionalDatetimeParameter;
 import static google.registry.request.RequestParameters.extractOptionalIntParameter;
 import static google.registry.request.RequestParameters.extractOptionalParameter;
 import static google.registry.request.RequestParameters.extractRequiredDatetimeParameter;
@@ -106,6 +107,13 @@ public class BatchModule {
     return extractIntParameter(req, RelockDomainAction.PREVIOUS_ATTEMPTS_PARAM);
   }
 
+  @Provides
+  @Parameter(ExpandRecurringBillingEventsAction.PARAM_CURSOR_TIME)
+  static Optional<DateTime> provideCursorTime(HttpServletRequest req) {
+    return extractOptionalDatetimeParameter(
+        req, ExpandRecurringBillingEventsAction.PARAM_CURSOR_TIME);
+  }
+
   @Provides
   @Named(QUEUE_ASYNC_ACTIONS)
   static Queue provideAsyncActionsPushQueue() {

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

@@ -311,7 +311,7 @@ public class DeleteContactsAndHostsAction implements Runnable {
     @Override
     public void reduce(final DeletionRequest deletionRequest, ReducerInput<Boolean> values) {
       final boolean hasNoActiveReferences = !Iterators.contains(values, true);
-      logger.atInfo().log("Processing async deletion request for %s", deletionRequest.key());
+      logger.atInfo().log("Processing async deletion request for %s.", deletionRequest.key());
       DeletionResult result =
           tm()
               .transactNew(
@@ -605,12 +605,12 @@ public class DeleteContactsAndHostsAction implements Runnable {
   static boolean doesResourceStateAllowDeletion(EppResource resource, DateTime now) {
     Key<EppResource> key = Key.create(resource);
     if (isDeleted(resource, now)) {
-      logger.atWarning().log("Cannot asynchronously delete %s because it is already deleted", key);
+      logger.atWarning().log("Cannot asynchronously delete %s because it is already deleted.", key);
       return false;
     }
     if (!resource.getStatusValues().contains(PENDING_DELETE)) {
       logger.atWarning().log(
-          "Cannot asynchronously delete %s because it is not in PENDING_DELETE", key);
+          "Cannot asynchronously delete %s because it is not in PENDING_DELETE.", key);
       return false;
     }
     return true;

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

@@ -167,7 +167,7 @@ public class DeleteExpiredDomainsAction implements Runnable {
 
   /** Runs the actual domain delete flow and returns whether the deletion was successful. */
   private boolean runDomainDeleteFlow(DomainBase domain) {
-    logger.atInfo().log("Attempting to delete domain %s", domain.getDomainName());
+    logger.atInfo().log("Attempting to delete domain '%s'.", domain.getDomainName());
     // Create a new transaction that the flow's execution will be enlisted in that loads the domain
     // transactionally. This way we can ensure that nothing else has modified the domain in question
     // in the intervening period since the query above found it.
@@ -203,7 +203,7 @@ public class DeleteExpiredDomainsAction implements Runnable {
 
     if (eppOutput.isPresent()) {
       if (eppOutput.get().isSuccess()) {
-        logger.atInfo().log("Successfully deleted domain %s", domain.getDomainName());
+        logger.atInfo().log("Successfully deleted domain '%s'.", domain.getDomainName());
       } else {
         logger.atSevere().log(
             "Failed to delete domain %s; EPP response:\n\n%s",

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

@@ -142,7 +142,7 @@ public class DeleteLoadTestDataAction implements Runnable {
     // that are linked to domains (since it would break the foreign keys)
     if (EppResourceUtils.isLinked(contact.createVKey(), clock.nowUtc())) {
       logger.atWarning().log(
-          "Cannot delete contact with repo ID %s since it is referenced from a domain",
+          "Cannot delete contact with repo ID %s since it is referenced from a domain.",
           contact.getRepoId());
       return;
     }
@@ -177,7 +177,7 @@ public class DeleteLoadTestDataAction implements Runnable {
         HistoryEntryDao.loadHistoryObjectsForResource(eppResource.createVKey());
     if (isDryRun) {
       logger.atInfo().log(
-          "Would delete repo ID %s along with %d history objects",
+          "Would delete repo ID %s along with %d history objects.",
           eppResource.getRepoId(), historyObjects.size());
     } else {
       historyObjects.forEach(tm()::delete);

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

@@ -228,7 +228,7 @@ public class DeleteProberDataAction implements Runnable {
     if (EppResourceUtils.isActive(domain, tm().getTransactionTime())) {
       if (isDryRun) {
         logger.atInfo().log(
-            "Would soft-delete the active domain: %s (%s)",
+            "Would soft-delete the active domain: %s (%s).",
             domain.getDomainName(), domain.getRepoId());
       } else {
         softDeleteDomain(domain, registryAdminRegistrarId, dnsQueue);
@@ -237,7 +237,7 @@ public class DeleteProberDataAction implements Runnable {
     } else {
       if (isDryRun) {
         logger.atInfo().log(
-            "Would hard-delete the non-active domain: %s (%s) and its dependents",
+            "Would hard-delete the non-active domain: %s (%s) and its dependents.",
             domain.getDomainName(), domain.getRepoId());
       } else {
         domainRepoIdsToHardDelete.add(domain.getRepoId());
@@ -331,7 +331,7 @@ public class DeleteProberDataAction implements Runnable {
           getContext().incrementCounter("skipped, non-prober data");
         }
       } catch (Throwable t) {
-        logger.atSevere().withCause(t).log("Error while deleting prober data for key %s", key);
+        logger.atSevere().withCause(t).log("Error while deleting prober data for key %s.", key);
         getContext().incrementCounter(String.format("error, kind %s", key.getKind()));
       }
     }
@@ -372,7 +372,7 @@ public class DeleteProberDataAction implements Runnable {
       if (EppResourceUtils.isActive(domain, now)) {
         if (isDryRun) {
           logger.atInfo().log(
-              "Would soft-delete the active domain: %s (%s)", domainName, domainKey);
+              "Would soft-delete the active domain: %s (%s).", domainName, domainKey);
         } else {
           tm().transact(() -> softDeleteDomain(domain, registryAdminRegistrarId, dnsQueue));
         }

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

@@ -148,9 +148,9 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
               .reduce(0, Integer::sum);
 
       if (!isDryRun) {
-        logger.atInfo().log("Saved OneTime billing events", numBillingEventsSaved);
+        logger.atInfo().log("Saved OneTime billing events.", numBillingEventsSaved);
       } else {
-        logger.atInfo().log("Generated OneTime billing events (dry run)", numBillingEventsSaved);
+        logger.atInfo().log("Generated OneTime billing events (dry run).", numBillingEventsSaved);
       }
       logger.atInfo().log(
           "Recurring event expansion %s complete for billing event range [%s, %s).",

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

@@ -173,7 +173,7 @@ public class RefreshDnsOnHostRenameAction implements Runnable {
                     retrier.callWithRetry(
                         () -> dnsQueue.addDomainRefreshTask(domainName),
                         TransientFailureException.class);
-                    logger.atInfo().log("Enqueued DNS refresh for domain %s.", domainName);
+                    logger.atInfo().log("Enqueued DNS refresh for domain '%s'.", domainName);
                   });
           deleteTasksWithRetry(
               refreshRequests,

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

@@ -313,7 +313,7 @@ public class RelockDomainAction implements Runnable {
         builder.add(new InternetAddress(registryLockEmailAddress));
       } catch (AddressException e) {
         // This shouldn't stop any other emails going out, so swallow it
-        logger.atWarning().log("Invalid email address %s", registryLockEmailAddress);
+        logger.atWarning().log("Invalid email address '%s'.", registryLockEmailAddress);
       }
     }
     return builder.build();

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

@@ -0,0 +1,134 @@
+// Copyright 2021 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.batch;
+
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static org.apache.http.HttpStatus.SC_INTERNAL_SERVER_ERROR;
+import static org.apache.http.HttpStatus.SC_OK;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.flogger.FluentLogger;
+import com.google.common.net.MediaType;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.model.contact.ContactHistory;
+import google.registry.request.Action;
+import google.registry.request.Action.Service;
+import google.registry.request.Response;
+import google.registry.request.auth.Auth;
+import google.registry.util.Clock;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+import javax.inject.Inject;
+import org.joda.time.DateTime;
+
+/**
+ * An action that wipes out Personal Identifiable Information (PII) fields of {@link ContactHistory}
+ * entities.
+ *
+ * <p>ContactHistory entities should be retained in the database for only certain amount of time.
+ * This periodic wipe out action only applies to SQL.
+ */
+@Action(
+    service = Service.BACKEND,
+    path = WipeOutContactHistoryPiiAction.PATH,
+    auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+public class WipeOutContactHistoryPiiAction implements Runnable {
+
+  public static final String PATH = "/_dr/task/wipeOutContactHistoryPii";
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private final Clock clock;
+  private final Response response;
+  private final int minMonthsBeforeWipeOut;
+  private final int wipeOutQueryBatchSize;
+
+  @Inject
+  public WipeOutContactHistoryPiiAction(
+      Clock clock,
+      @Config("minMonthsBeforeWipeOut") int minMonthsBeforeWipeOut,
+      @Config("wipeOutQueryBatchSize") int wipeOutQueryBatchSize,
+      Response response) {
+    this.clock = clock;
+    this.response = response;
+    this.minMonthsBeforeWipeOut = minMonthsBeforeWipeOut;
+    this.wipeOutQueryBatchSize = wipeOutQueryBatchSize;
+  }
+
+  @Override
+  public void run() {
+    response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
+    try {
+      int totalNumOfWipedEntities = 0;
+      DateTime wipeOutTime = clock.nowUtc().minusMonths(minMonthsBeforeWipeOut);
+      logger.atInfo().log(
+          "About to wipe out all PII of contact history entities prior to %s.", wipeOutTime);
+
+      int numOfWipedEntities = 0;
+      do {
+        numOfWipedEntities =
+            jpaTm()
+                .transact(
+                    () ->
+                        wipeOutContactHistoryData(
+                            getNextContactHistoryEntitiesWithPiiBatch(wipeOutTime)));
+        totalNumOfWipedEntities += numOfWipedEntities;
+      } while (numOfWipedEntities > 0);
+      logger.atInfo().log(
+          "Wiped out PII of %d ContactHistory entities in total.", totalNumOfWipedEntities);
+      response.setStatus(SC_OK);
+
+    } catch (Exception e) {
+      logger.atSevere().withCause(e).log(
+          "Exception thrown during the process of wiping out contact history PII.");
+      response.setStatus(SC_INTERNAL_SERVER_ERROR);
+      response.setPayload(
+          String.format(
+              "Exception thrown during the process of wiping out contact history PII with cause"
+                  + ": %s",
+              e));
+    }
+  }
+
+  /**
+   * Returns a stream of up to {@link #wipeOutQueryBatchSize} {@link ContactHistory} entities
+   * containing PII that are prior to @param wipeOutTime.
+   */
+  @VisibleForTesting
+  Stream<ContactHistory> getNextContactHistoryEntitiesWithPiiBatch(DateTime wipeOutTime) {
+    // email is one of the required fields in EPP, meaning it's initially not null.
+    // Therefore, checking if it's null is one way to avoid processing contact history entities
+    // that have been processed previously. Refer to RFC 5733 for more information.
+    return jpaTm()
+        .query(
+            "FROM ContactHistory WHERE modificationTime < :wipeOutTime " + "AND email IS NOT NULL",
+            ContactHistory.class)
+        .setParameter("wipeOutTime", wipeOutTime)
+        .setMaxResults(wipeOutQueryBatchSize)
+        .getResultStream();
+  }
+
+  /** Wipes out the PII of each of the {@link ContactHistory} entities in the stream. */
+  @VisibleForTesting
+  int wipeOutContactHistoryData(Stream<ContactHistory> contactHistoryEntities) {
+    AtomicInteger numOfEntities = new AtomicInteger(0);
+    contactHistoryEntities.forEach(
+        contactHistoryEntity -> {
+          jpaTm().update(contactHistoryEntity.asBuilder().wipeOutPii().build());
+          numOfEntities.incrementAndGet();
+        });
+    logger.atInfo().log(
+        "Wiped out all PII fields of %d ContactHistory entities.", numOfEntities.get());
+    return numOfEntities.get();
+  }
+}

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

@@ -17,7 +17,6 @@ package google.registry.beam.common;
 import static com.google.common.base.Verify.verify;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 
-import google.registry.backup.AppEngineEnvironment;
 import google.registry.model.contact.ContactResource;
 import google.registry.persistence.transaction.CriteriaQueryBuilder;
 import google.registry.persistence.transaction.JpaTransactionManager;
@@ -59,18 +58,16 @@ public class JpaDemoPipeline implements Serializable {
                   public void processElement() {
                     // AppEngineEnvironment is needed as long as JPA entity classes still depends
                     // on Objectify.
-                    try (AppEngineEnvironment allowOfyEntity = new AppEngineEnvironment()) {
-                      int result =
-                          (Integer)
-                              jpaTm()
-                                  .transact(
-                                      () ->
-                                          jpaTm()
-                                              .getEntityManager()
-                                              .createNativeQuery("select 1;")
-                                              .getSingleResult());
-                      verify(result == 1, "Expecting 1, got %s.", result);
-                    }
+                    int result =
+                        (Integer)
+                            jpaTm()
+                                .transact(
+                                    () ->
+                                        jpaTm()
+                                            .getEntityManager()
+                                            .createNativeQuery("select 1;")
+                                            .getSingleResult());
+                    verify(result == 1, "Expecting 1, got %s.", result);
                     counter.inc();
                   }
                 }));

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

@@ -20,11 +20,9 @@ import static org.apache.beam.sdk.values.TypeDescriptors.integers;
 import com.google.auto.value.AutoValue;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Streams;
-import google.registry.backup.AppEngineEnvironment;
 import google.registry.beam.common.RegistryQuery.CriteriaQuerySupplier;
 import google.registry.model.UpdateAutoTimestamp;
 import google.registry.model.UpdateAutoTimestamp.DisableAutoUpdateResource;
-import google.registry.model.ofy.ObjectifyService;
 import google.registry.model.replay.SqlEntity;
 import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.persistence.transaction.TransactionManagerFactory;
@@ -211,14 +209,9 @@ public final class RegistryJpaIO {
 
       @ProcessElement
       public void processElement(OutputReceiver<T> outputReceiver) {
-        // AppEngineEnvironment is need for handling VKeys, which involve Ofy keys. Unlike
-        // SqlBatchWriter, it is unnecessary to initialize ObjectifyService in this class.
-        try (AppEngineEnvironment env = new AppEngineEnvironment()) {
-          // TODO(b/187210388): JpaTransactionManager should support non-transactional query.
-          jpaTm()
-              .transactNoRetry(
-                  () -> query.stream().map(resultMapper::apply).forEach(outputReceiver::output));
-        }
+        jpaTm()
+            .transactNoRetry(
+                () -> query.stream().map(resultMapper::apply).forEach(outputReceiver::output));
       }
     }
   }
@@ -364,16 +357,6 @@ public final class RegistryJpaIO {
       this.withAutoTimestamp = withAutoTimestamp;
     }
 
-    @Setup
-    public void setup() {
-      // AppEngineEnvironment is needed as long as Objectify keys are still involved in the handling
-      // of SQL entities (e.g., in VKeys). ObjectifyService needs to be initialized when conversion
-      // between Ofy entity and Datastore entity is needed.
-      try (AppEngineEnvironment env = new AppEngineEnvironment()) {
-        ObjectifyService.initOfy();
-      }
-    }
-
     @ProcessElement
     public void processElement(@Element KV<ShardedKey<Integer>, Iterable<T>> kv) {
       if (withAutoTimestamp) {
@@ -386,19 +369,17 @@ public final class RegistryJpaIO {
     }
 
     private void actuallyProcessElement(@Element KV<ShardedKey<Integer>, Iterable<T>> kv) {
-      try (AppEngineEnvironment env = new AppEngineEnvironment()) {
-        ImmutableList<Object> entities =
-            Streams.stream(kv.getValue())
-                .map(this.jpaConverter::apply)
-                // TODO(b/177340730): post migration delete the line below.
-                .filter(Objects::nonNull)
-                .collect(ImmutableList.toImmutableList());
-        try {
-          jpaTm().transact(() -> jpaTm().putAll(entities));
-          counter.inc(entities.size());
-        } catch (RuntimeException e) {
-          processSingly(entities);
-        }
+      ImmutableList<Object> entities =
+          Streams.stream(kv.getValue())
+              .map(this.jpaConverter::apply)
+              // TODO(b/177340730): post migration delete the line below.
+              .filter(Objects::nonNull)
+              .collect(ImmutableList.toImmutableList());
+      try {
+        jpaTm().transact(() -> jpaTm().putAll(entities));
+        counter.inc(entities.size());
+      } catch (RuntimeException e) {
+        processSingly(entities);
       }
     }
 

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

@@ -20,6 +20,8 @@ import com.google.auto.service.AutoService;
 import com.google.common.flogger.FluentLogger;
 import dagger.Lazy;
 import google.registry.config.RegistryEnvironment;
+import google.registry.config.SystemPropertySetter;
+import google.registry.model.AppEngineEnvironment;
 import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.persistence.transaction.TransactionManagerFactory;
 import org.apache.beam.sdk.harness.JvmInitializer;
@@ -35,18 +37,26 @@ import org.apache.beam.sdk.options.PipelineOptions;
 @AutoService(JvmInitializer.class)
 public class RegistryPipelineWorkerInitializer implements JvmInitializer {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  public static final String PROPERTY = "google.registry.beam";
 
   @Override
   public void beforeProcessing(PipelineOptions options) {
     RegistryPipelineOptions registryOptions = options.as(RegistryPipelineOptions.class);
     RegistryEnvironment environment = registryOptions.getRegistryEnvironment();
     if (environment == null || environment.equals(RegistryEnvironment.UNITTEST)) {
-      return;
+      throw new RuntimeException(
+          "A registry environment must be specified in the pipeline options.");
     }
-    logger.atInfo().log("Setting up RegistryEnvironment: %s", environment);
+    logger.atInfo().log("Setting up RegistryEnvironment %s.", environment);
     environment.setup();
     Lazy<JpaTransactionManager> transactionManagerLazy =
         toRegistryPipelineComponent(registryOptions).getJpaTransactionManager();
     TransactionManagerFactory.setJpaTmOnBeamWorker(transactionManagerLazy::get);
+    // Masquarade all threads as App Engine threads so we can create Ofy keys in the pipeline. Also
+    // loads all ofy entities.
+    new AppEngineEnvironment("Beam").setEnvironmentForAllThreads();
+    // Set the system property so that we can call IdService.allocateId() without access to
+    // datastore.
+    SystemPropertySetter.PRODUCTION_IMPL.setProperty(PROPERTY, "true");
   }
 }

core/src/main/java/google/registry/beam/datastore/DatastoreV1.java

@@ -169,14 +169,15 @@ public class DatastoreV1 {
       int numSplits;
       try {
         long estimatedSizeBytes = getEstimatedSizeBytes(datastore, query, namespace);
-        logger.atInfo().log("Estimated size bytes for the query is: %s", estimatedSizeBytes);
+        logger.atInfo().log("Estimated size for the query is %d bytes.", estimatedSizeBytes);
         numSplits =
             (int)
                 Math.min(
                     NUM_QUERY_SPLITS_MAX,
                     Math.round(((double) estimatedSizeBytes) / DEFAULT_BUNDLE_SIZE_BYTES));
       } catch (Exception e) {
-        logger.atWarning().log("Failed the fetch estimatedSizeBytes for query: %s", query, e);
+        logger.atWarning().withCause(e).log(
+            "Failed the fetch estimatedSizeBytes for query: %s", query);
         // Fallback in case estimated size is unavailable.
         numSplits = NUM_QUERY_SPLITS_MIN;
       }
@@ -215,7 +216,7 @@ public class DatastoreV1 {
     private static Entity getLatestTableStats(
         String ourKind, @Nullable String namespace, Datastore datastore) throws DatastoreException {
       long latestTimestamp = queryLatestStatisticsTimestamp(datastore, namespace);
-      logger.atInfo().log("Latest stats timestamp for kind %s is %s", ourKind, latestTimestamp);
+      logger.atInfo().log("Latest stats timestamp for kind %s is %s.", ourKind, latestTimestamp);
 
       Query.Builder queryBuilder = Query.newBuilder();
       if (Strings.isNullOrEmpty(namespace)) {
@@ -234,7 +235,7 @@ public class DatastoreV1 {
       long now = System.currentTimeMillis();
       RunQueryResponse response = datastore.runQuery(request);
       logger.atFine().log(
-          "Query for per-kind statistics took %sms", System.currentTimeMillis() - now);
+          "Query for per-kind statistics took %d ms.", System.currentTimeMillis() - now);
 
       QueryResultBatch batch = response.getBatch();
       if (batch.getEntityResultsCount() == 0) {
@@ -330,7 +331,7 @@ public class DatastoreV1 {
           logger.atWarning().log(
               "Failed to translate Gql query '%s': %s", gqlQueryWithZeroLimit, e.getMessage());
           logger.atWarning().log(
-              "User query might have a limit already set, so trying without zero limit");
+              "User query might have a limit already set, so trying without zero limit.");
           // Retry without the zero limit.
           return translateGqlQuery(gql, datastore, namespace);
         } else {
@@ -514,10 +515,10 @@ public class DatastoreV1 {
       @ProcessElement
       public void processElement(ProcessContext c) throws Exception {
         String gqlQuery = c.element();
-        logger.atInfo().log("User query: '%s'", gqlQuery);
+        logger.atInfo().log("User query: '%s'.", gqlQuery);
         Query query =
             translateGqlQueryWithLimitCheck(gqlQuery, datastore, v1Options.getNamespace());
-        logger.atInfo().log("User gql query translated to Query(%s)", query);
+        logger.atInfo().log("User gql query translated to Query(%s).", query);
         c.output(query);
       }
     }
@@ -573,7 +574,7 @@ public class DatastoreV1 {
           estimatedNumSplits = numSplits;
         }
 
-        logger.atInfo().log("Splitting the query into %s splits", estimatedNumSplits);
+        logger.atInfo().log("Splitting the query into %d splits.", estimatedNumSplits);
         List<Query> querySplits;
         try {
           querySplits =
@@ -647,7 +648,7 @@ public class DatastoreV1 {
               throw exception;
             }
             if (!BackOffUtils.next(sleeper, backoff)) {
-              logger.atSevere().log("Aborting after %s retries.", MAX_RETRIES);
+              logger.atSevere().log("Aborting after %d retries.", MAX_RETRIES);
               throw exception;
             }
           }

core/src/main/java/google/registry/beam/initsql/InitSqlPipeline.java

@@ -20,11 +20,11 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
-import google.registry.backup.AppEngineEnvironment;
 import google.registry.backup.VersionedEntity;
 import google.registry.beam.common.RegistryJpaIO;
 import google.registry.beam.initsql.Transforms.RemoveDomainBaseForeignKeys;
 import google.registry.model.billing.BillingEvent;
+import google.registry.model.common.Cursor;
 import google.registry.model.contact.ContactResource;
 import google.registry.model.domain.DomainBase;
 import google.registry.model.domain.token.AllocationToken;
@@ -62,6 +62,7 @@ import org.joda.time.DateTime;
  * <ol>
  *   <li>{@link Registry}: Assumes that {@code PremiumList} and {@code ReservedList} have been set
  *       up in the SQL database.
+ *   <li>{@link Cursor}: Logically can depend on {@code Registry}, but without foreign key.
  *   <li>{@link Registrar}: Logically depends on {@code Registry}, Foreign key not modeled yet.
  *   <li>{@link ContactResource}: references {@code Registrar}
  *   <li>{@link RegistrarContact}: references {@code Registrar}.
@@ -101,7 +102,11 @@ public class InitSqlPipeline implements Serializable {
    */
   private static final ImmutableList<Class<?>> PHASE_ONE_ORDERED =
       ImmutableList.of(
-          Registry.class, Registrar.class, ContactResource.class, RegistrarContact.class);
+          Registry.class,
+          Cursor.class,
+          Registrar.class,
+          ContactResource.class,
+          RegistrarContact.class);
 
   /**
    * Datastore kinds to be written to the SQL database after the cleansed version of {@link
@@ -224,9 +229,7 @@ public class InitSqlPipeline implements Serializable {
   }
 
   private static ImmutableList<String> toKindStrings(Collection<Class<?>> entityClasses) {
-    try (AppEngineEnvironment env = new AppEngineEnvironment()) {
-      return entityClasses.stream().map(Key::getKind).collect(ImmutableList.toImmutableList());
-    }
+    return entityClasses.stream().map(Key::getKind).collect(ImmutableList.toImmutableList());
   }
 
   public static void main(String[] args) {

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

@@ -172,7 +172,7 @@ public class RdeIO {
 
       // Write a gigantic XML file to GCS. We'll start by opening encrypted out/err file handles.
 
-      logger.atInfo().log("Writing %s and %s", xmlFilename, xmlLengthFilename);
+      logger.atInfo().log("Writing files '%s' and '%s'.", xmlFilename, xmlLengthFilename);
       try (OutputStream gcsOutput = gcsUtils.openOutputStream(xmlFilename);
           OutputStream lengthOutput = gcsUtils.openOutputStream(xmlLengthFilename);
           OutputStream ghostrydeEncoder = Ghostryde.encoder(gcsOutput, stagingKey, lengthOutput);
@@ -219,7 +219,7 @@ public class RdeIO {
       //
       // This will be sent to ICANN once we're done uploading the big XML to the escrow provider.
       if (mode == RdeMode.FULL) {
-        logger.atInfo().log("Writing %s", reportFilename);
+        logger.atInfo().log("Writing file '%s'.", reportFilename);
         try (OutputStream gcsOutput = gcsUtils.openOutputStream(reportFilename);
             OutputStream ghostrydeEncoder = Ghostryde.encoder(gcsOutput, stagingKey)) {
           counter.makeReport(id, watermark, header, revision).marshal(ghostrydeEncoder, UTF_8);
@@ -229,7 +229,7 @@ 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");
+        logger.atInfo().log("Manual operation; not advancing cursor or enqueuing upload task.");
       } else {
         outputReceiver.output(KV.of(key, revision));
       }
@@ -265,7 +265,7 @@ public class RdeIO {
                     key);
                 tm().put(Cursor.create(key.cursor(), newPosition, registry));
                 logger.atInfo().log(
-                    "Rolled forward %s on %s cursor to %s", key.cursor(), key.tld(), newPosition);
+                    "Rolled forward %s on %s cursor to %s.", key.cursor(), key.tld(), newPosition);
                 RdeRevision.saveRevision(key.tld(), key.watermark(), key.mode(), revision);
               });
     }

core/src/main/java/google/registry/beam/spec11/SafeBrowsingTransforms.java

@@ -218,7 +218,7 @@ public class SafeBrowsingTransforms {
         throws JSONException, IOException {
       int statusCode = response.getStatusLine().getStatusCode();
       if (statusCode != SC_OK) {
-        logger.atWarning().log("Got unexpected status code %s from response", statusCode);
+        logger.atWarning().log("Got unexpected status code %s from response.", statusCode);
       } else {
         // Unpack the response body
         JSONObject responseBody =
@@ -227,7 +227,7 @@ public class SafeBrowsingTransforms {
                     new InputStreamReader(response.getEntity().getContent(), UTF_8)));
         logger.atInfo().log("Got response: %s", responseBody.toString());
         if (responseBody.length() == 0) {
-          logger.atInfo().log("Response was empty, no threats detected");
+          logger.atInfo().log("Response was empty, no threats detected.");
         } else {
           // Emit all DomainNameInfos with their API results.
           JSONArray threatMatches = responseBody.getJSONArray("matches");

core/src/main/java/google/registry/beam/spec11/Spec11Pipeline.java

@@ -16,6 +16,7 @@ package google.registry.beam.spec11;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static google.registry.beam.BeamUtils.getQueryFromFile;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 
 import com.google.auto.value.AutoValue;
 import com.google.common.collect.ImmutableSet;
@@ -30,6 +31,7 @@ import google.registry.model.domain.DomainBase;
 import google.registry.model.reporting.Spec11ThreatMatch;
 import google.registry.model.reporting.Spec11ThreatMatch.ThreatType;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
+import google.registry.persistence.VKey;
 import google.registry.util.Retrier;
 import google.registry.util.SqlTemplate;
 import google.registry.util.UtilsModule;
@@ -41,6 +43,7 @@ import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.io.TextIO;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.GroupByKey;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.ParDo;
@@ -113,15 +116,43 @@ public class Spec11Pipeline implements Serializable {
   }
 
   static PCollection<DomainNameInfo> readFromCloudSql(Pipeline pipeline) {
-    Read<Object[], DomainNameInfo> read =
+    Read<Object[], KV<String, String>> read =
         RegistryJpaIO.read(
-            "select d, r.emailAddress from Domain d join Registrar r on"
-                + " d.currentSponsorClientId = r.clientIdentifier where r.type = 'REAL'"
-                + " and d.deletionTime > now()",
+            "select d.repoId, r.emailAddress from Domain d join Registrar r on"
+                + " d.currentSponsorClientId = r.clientIdentifier where r.type = 'REAL' and"
+                + " d.deletionTime > now()",
             false,
             Spec11Pipeline::parseRow);
 
-    return pipeline.apply("Read active domains from Cloud SQL", read);
+    return pipeline
+        .apply("Read active domains from Cloud SQL", read)
+        .apply(
+            "Build DomainNameInfo",
+            ParDo.of(
+                new DoFn<KV<String, String>, DomainNameInfo>() {
+                  @ProcessElement
+                  public void processElement(
+                      @Element KV<String, String> input, OutputReceiver<DomainNameInfo> output) {
+                    DomainBase domainBase =
+                        jpaTm()
+                            .transact(
+                                () ->
+                                    jpaTm()
+                                        .loadByKey(
+                                            VKey.createSql(DomainBase.class, input.getKey())));
+                    String emailAddress = input.getValue();
+                    if (emailAddress == null) {
+                      emailAddress = "";
+                    }
+                    DomainNameInfo domainNameInfo =
+                        DomainNameInfo.create(
+                            domainBase.getDomainName(),
+                            domainBase.getRepoId(),
+                            domainBase.getCurrentSponsorRegistrarId(),
+                            emailAddress);
+                    output.output(domainNameInfo);
+                  }
+                }));
   }
 
   static PCollection<DomainNameInfo> readFromBigQuery(
@@ -142,17 +173,8 @@ public class Spec11Pipeline implements Serializable {
             .withTemplateCompatibility());
   }
 
-  private static DomainNameInfo parseRow(Object[] row) {
-    DomainBase domainBase = (DomainBase) row[0];
-    String emailAddress = (String) row[1];
-    if (emailAddress == null) {
-      emailAddress = "";
-    }
-    return DomainNameInfo.create(
-        domainBase.getDomainName(),
-        domainBase.getRepoId(),
-        domainBase.getCurrentSponsorRegistrarId(),
-        emailAddress);
+  private static KV<String, String> parseRow(Object[] row) {
+    return KV.of((String) row[0], (String) row[1]);
   }
 
   static void saveToSql(

core/src/main/java/google/registry/bigquery/BigqueryConnection.java

@@ -632,7 +632,7 @@ public class BigqueryConnection implements AutoCloseable {
   private static String summarizeCompletedJob(Job job) {
     JobStatistics stats = job.getStatistics();
     return String.format(
-        "Job took %,.3f seconds after a %,.3f second delay and processed %,d bytes (%s)",
+        "Job took %,.3f seconds after a %,.3f second delay and processed %,d bytes (%s).",
         (stats.getEndTime() - stats.getStartTime()) / 1000.0,
         (stats.getStartTime() - stats.getCreationTime()) / 1000.0,
         stats.getTotalBytesProcessed(),
@@ -706,17 +706,17 @@ public class BigqueryConnection implements AutoCloseable {
   }
 
   /**
-   * Helper that creates a dataset with this name if it doesn't already exist, and returns true
-   * if creation took place.
+   * Helper that creates a dataset with this name if it doesn't already exist, and returns true if
+   * creation took place.
    */
-  public boolean createDatasetIfNeeded(String datasetName) throws IOException {
+  private boolean createDatasetIfNeeded(String datasetName) throws IOException {
     if (!checkDatasetExists(datasetName)) {
       bigquery.datasets()
           .insert(getProjectId(), new Dataset().setDatasetReference(new DatasetReference()
               .setProjectId(getProjectId())
               .setDatasetId(datasetName)))
           .execute();
-      logger.atInfo().log("Created dataset: %s:%s\n", getProjectId(), datasetName);
+      logger.atInfo().log("Created dataset: %s: %s.", getProjectId(), datasetName);
       return true;
     }
     return false;
@@ -732,9 +732,8 @@ public class BigqueryConnection implements AutoCloseable {
                   .setDefaultDataset(getDataset())
                   .setDestinationTable(table))));
     } catch (BigqueryJobFailureException e) {
-      if (e.getReason().equals("duplicate")) {
-        // Table already exists.
-      } else {
+      if (!e.getReason().equals("duplicate")) {
+        // Throw if it failed for any reason other than table already existing.
         throw e;
       }
     }

core/src/main/java/google/registry/bigquery/CheckedBigquery.java

@@ -116,7 +116,7 @@ public class CheckedBigquery {
           .setTableReference(table))
           .execute();
       logger.atInfo().log(
-          "Created BigQuery table %s:%s.%s",
+          "Created BigQuery table %s:%s.%s.",
           table.getProjectId(), table.getDatasetId(), table.getTableId());
     } catch (IOException e) {
       // Swallow errors about a table that exists, and throw any other ones.

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

@@ -1306,6 +1306,18 @@ public final class RegistryConfig {
     public static ImmutableSet<String> provideAllowedEcdsaCurves(RegistryConfigSettings config) {
       return ImmutableSet.copyOf(config.sslCertificateValidation.allowedEcdsaCurves);
     }
+
+    @Provides
+    @Config("minMonthsBeforeWipeOut")
+    public static int provideMinMonthsBeforeWipeOut(RegistryConfigSettings config) {
+      return config.contactHistory.minMonthsBeforeWipeOut;
+    }
+
+    @Provides
+    @Config("wipeOutQueryBatchSize")
+    public static int provideWipeOutQueryBatchSize(RegistryConfigSettings config) {
+      return config.contactHistory.wipeOutQueryBatchSize;
+    }
   }
 
   /** Returns the App Engine project ID, which is based off the environment name. */

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

@@ -41,6 +41,7 @@ public class RegistryConfigSettings {
   public Keyring keyring;
   public RegistryTool registryTool;
   public SslCertificateValidation sslCertificateValidation;
+  public ContactHistory contactHistory;
 
   /** Configuration options that apply to the entire App Engine project. */
   public static class AppEngine {
@@ -234,4 +235,10 @@ public class RegistryConfigSettings {
     public String expirationWarningEmailBodyText;
     public String expirationWarningEmailSubjectText;
   }
+
+  /** Configuration for contact history. */
+  public static class ContactHistory {
+    public int minMonthsBeforeWipeOut;
+    public int wipeOutQueryBatchSize;
+  }
 }

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

@@ -442,6 +442,13 @@ registryTool:
   # OAuth client secret used by the tool.
   clientSecret: YOUR_CLIENT_SECRET
 
+# Configuration options for handling contact history.
+contactHistory:
+  # The number of months that a ContactHistory entity should be stored in the database.
+  minMonthsBeforeWipeOut: 18
+  # The batch size for querying ContactHistory table in the database.
+  wipeOutQueryBatchSize: 500
+
 # Configuration options for checking SSL certificates.
 sslCertificateValidation:
   # A map specifying the maximum amount of days the certificate can be valid.

core/src/main/java/google/registry/cron/TldFanoutAction.java

@@ -141,7 +141,7 @@ public final class TldFanoutAction implements Runnable {
     StringBuilder outputPayload =
         new StringBuilder(
             String.format("OK: Launched the following %d tasks in queue %s\n", tlds.size(), queue));
-    logger.atInfo().log("Launching %d tasks in queue %s", tlds.size(), queue);
+    logger.atInfo().log("Launching %d tasks in queue %s.", tlds.size(), queue);
     if (tlds.isEmpty()) {
       logger.atWarning().log("No TLDs to fan-out!");
     }
@@ -153,7 +153,7 @@ public final class TldFanoutAction implements Runnable {
               "- Task: '%s', tld: '%s', endpoint: '%s'\n",
               createdTask.getName(), tld, createdTask.getAppEngineHttpRequest().getRelativeUri()));
       logger.atInfo().log(
-          "Task: '%s', tld: '%s', endpoint: '%s'",
+          "Task: '%s', tld: '%s', endpoint: '%s'.",
           createdTask.getName(), tld, createdTask.getAppEngineHttpRequest().getRelativeUri());
     }
     response.setContentType(PLAIN_TEXT_UTF_8);

core/src/main/java/google/registry/dns/DnsQueue.java

@@ -107,7 +107,7 @@ public class DnsQueue {
   private TaskHandle addToQueue(
       TargetType targetType, String targetName, String tld, Duration countdown) {
     logger.atInfo().log(
-        "Adding task type=%s, target=%s, tld=%s to pull queue %s (%d tasks currently on queue)",
+        "Adding task type=%s, target=%s, tld=%s to pull queue %s (%d tasks currently on queue).",
         targetType, targetName, tld, DNS_PULL_QUEUE_NAME, queue.fetchStatistics().getNumTasks());
     return queue.add(
         TaskOptions.Builder.withDefaults()
@@ -166,7 +166,7 @@ public class DnsQueue {
           "There are %d tasks in the DNS queue '%s'.", numTasks, DNS_PULL_QUEUE_NAME);
       return queue.leaseTasks(leaseDuration.getMillis(), MILLISECONDS, leaseTasksBatchSize);
     } catch (TransientFailureException | DeadlineExceededException e) {
-      logger.atSevere().withCause(e).log("Failed leasing tasks too fast");
+      logger.atSevere().withCause(e).log("Failed leasing tasks too fast.");
       return ImmutableList.of();
     }
   }
@@ -176,7 +176,7 @@ public class DnsQueue {
     try {
       queue.deleteTask(tasks);
     } catch (TransientFailureException | DeadlineExceededException e) {
-      logger.atSevere().withCause(e).log("Failed deleting tasks too fast");
+      logger.atSevere().withCause(e).log("Failed deleting tasks too fast.");
     }
   }
 }

core/src/main/java/google/registry/dns/PublishDnsUpdatesAction.java

@@ -104,7 +104,7 @@ public final class PublishDnsUpdatesAction implements Runnable, Callable<Void> {
         new Duration(enqueuedTime, now));
     logger.atInfo().log(
         "publishDnsWriter latency statistics: TLD: %s, dnsWriter: %s, actionStatus: %s, "
-            + "numItems: %d, timeSinceCreation: %s, timeInQueue: %s",
+            + "numItems: %d, timeSinceCreation: %s, timeInQueue: %s.",
         tld,
         dnsWriter,
         status,
@@ -144,7 +144,7 @@ public final class PublishDnsUpdatesAction implements Runnable, Callable<Void> {
 
   /** Adds all the domains and hosts in the batch back to the queue to be processed later. */
   private void requeueBatch() {
-    logger.atInfo().log("Requeueing batch for retry");
+    logger.atInfo().log("Requeueing batch for retry.");
     for (String domain : nullToEmpty(domains)) {
       dnsQueue.addDomainRefreshTask(domain);
     }
@@ -158,14 +158,14 @@ public final class PublishDnsUpdatesAction implements Runnable, Callable<Void> {
     // LockIndex should always be within [1, numPublishLocks]
     if (lockIndex > numPublishLocks || lockIndex <= 0) {
       logger.atSevere().log(
-          "Lock index should be within [1,%d], got %d instead", numPublishLocks, lockIndex);
+          "Lock index should be within [1,%d], got %d instead.", numPublishLocks, lockIndex);
       return false;
     }
     // Check if the Registry object's num locks has changed since this task was batched
     int registryNumPublishLocks = Registry.get(tld).getNumDnsPublishLocks();
     if (registryNumPublishLocks != numPublishLocks) {
       logger.atWarning().log(
-          "Registry numDnsPublishLocks %d out of sync with parameter %d",
+          "Registry numDnsPublishLocks %d out of sync with parameter %d.",
           registryNumPublishLocks, numPublishLocks);
       return false;
     }
@@ -179,7 +179,7 @@ public final class PublishDnsUpdatesAction implements Runnable, Callable<Void> {
     DnsWriter writer = dnsWriterProxy.getByClassNameForTld(dnsWriter, tld);
 
     if (writer == null) {
-      logger.atWarning().log("Couldn't get writer %s for TLD %s", dnsWriter, tld);
+      logger.atWarning().log("Couldn't get writer %s for TLD %s.", dnsWriter, tld);
       recordActionResult(ActionStatus.BAD_WRITER);
       requeueBatch();
       return;
@@ -190,11 +190,11 @@ public final class PublishDnsUpdatesAction implements Runnable, Callable<Void> {
     for (String domain : nullToEmpty(domains)) {
       if (!DomainNameUtils.isUnder(
           InternetDomainName.from(domain), InternetDomainName.from(tld))) {
-        logger.atSevere().log("%s: skipping domain %s not under tld", tld, domain);
+        logger.atSevere().log("%s: skipping domain %s not under TLD.", tld, domain);
         domainsRejected += 1;
       } else {
         writer.publishDomain(domain);
-        logger.atInfo().log("%s: published domain %s", tld, domain);
+        logger.atInfo().log("%s: published domain %s.", tld, domain);
         domainsPublished += 1;
       }
     }
@@ -206,11 +206,11 @@ public final class PublishDnsUpdatesAction implements Runnable, Callable<Void> {
     for (String host : nullToEmpty(hosts)) {
       if (!DomainNameUtils.isUnder(
           InternetDomainName.from(host), InternetDomainName.from(tld))) {
-        logger.atSevere().log("%s: skipping host %s not under tld", tld, host);
+        logger.atSevere().log("%s: skipping host %s not under TLD.", tld, host);
         hostsRejected += 1;
       } else {
         writer.publishHost(host);
-        logger.atInfo().log("%s: published host %s", tld, host);
+        logger.atInfo().log("%s: published host %s.", tld, host);
         hostsPublished += 1;
       }
     }
@@ -233,7 +233,7 @@ public final class PublishDnsUpdatesAction implements Runnable, Callable<Void> {
           tld, dnsWriter, commitStatus, duration, domainsPublished, hostsPublished);
       logger.atInfo().log(
           "writer.commit() statistics: TLD: %s, dnsWriter: %s, commitStatus: %s, duration: %s, "
-              + "domainsPublished: %d, domainsRejected: %d, hostsPublished: %d, hostsRejected: %d",
+              + "domainsPublished: %d, domainsRejected: %d, hostsPublished: %d, hostsRejected: %d.",
           tld,
           dnsWriter,
           commitStatus,

core/src/main/java/google/registry/dns/writer/clouddns/CloudDnsWriter.java

@@ -180,11 +180,11 @@ public class CloudDnsWriter extends BaseDnsWriter {
 
     desiredRecords.put(absoluteDomainName, domainRecords.build());
     logger.atFine().log(
-        "Will write %d records for domain %s", domainRecords.build().size(), absoluteDomainName);
+        "Will write %d records for domain '%s'.", domainRecords.build().size(), absoluteDomainName);
   }
 
   private void publishSubordinateHost(String hostName) {
-    logger.atInfo().log("Publishing glue records for %s", hostName);
+    logger.atInfo().log("Publishing glue records for host '%s'.", hostName);
     // Canonicalize name
     String absoluteHostName = getAbsoluteHostName(hostName);
 
@@ -250,7 +250,7 @@ public class CloudDnsWriter extends BaseDnsWriter {
 
     // Host not managed by our registry, no need to update DNS.
     if (!tld.isPresent()) {
-      logger.atSevere().log("publishHost called for invalid host %s", hostName);
+      logger.atSevere().log("publishHost called for invalid host '%s'.", hostName);
       return;
     }
 
@@ -273,7 +273,7 @@ public class CloudDnsWriter extends BaseDnsWriter {
     ImmutableMap<String, ImmutableSet<ResourceRecordSet>> desiredRecordsCopy =
         ImmutableMap.copyOf(desiredRecords);
     retrier.callWithRetry(() -> mutateZone(desiredRecordsCopy), ZoneStateException.class);
-    logger.atInfo().log("Wrote to Cloud DNS");
+    logger.atInfo().log("Wrote to Cloud DNS.");
   }
 
   /** Returns the glue records for in-bailiwick nameservers for the given domain+records. */
@@ -329,7 +329,7 @@ public class CloudDnsWriter extends BaseDnsWriter {
    */
   private Map<String, List<ResourceRecordSet>> getResourceRecordsForDomains(
       Set<String> domainNames) {
-    logger.atFine().log("Fetching records for %s", domainNames);
+    logger.atFine().log("Fetching records for domain '%s'.", domainNames);
     // As per Concurrent.transform() - if numThreads or domainNames.size() < 2, it will not use
     // threading.
     return ImmutableMap.copyOf(
@@ -381,11 +381,11 @@ public class CloudDnsWriter extends BaseDnsWriter {
     ImmutableSet<ResourceRecordSet> intersection =
         Sets.intersection(additions, deletions).immutableCopy();
     logger.atInfo().log(
-        "There are %d common items out of the %d items in 'additions' and %d items in 'deletions'",
+        "There are %d common items out of the %d items in 'additions' and %d items in 'deletions'.",
         intersection.size(), additions.size(), deletions.size());
     // Exit early if we have nothing to update - dnsConnection doesn't work on empty changes
     if (additions.equals(deletions)) {
-      logger.atInfo().log("Returning early because additions is the same as deletions");
+      logger.atInfo().log("Returning early because additions are the same as deletions.");
       return;
     }
     Change change =

core/src/main/java/google/registry/env/common/backend/WEB-INF/web.xml

@@ -391,6 +391,13 @@
     <url-pattern>/_dr/task/relockDomain</url-pattern>
   </servlet-mapping>
 
+  <!-- Background action to wipe out PII fields of ContactHistory entities that
+have been in the database for a certain period of time. -->
+  <servlet-mapping>
+    <servlet-name>backend-servlet</servlet-name>
+    <url-pattern>/_dr/task/wipeOutContactHistoryPii</url-pattern>
+  </servlet-mapping>
+
   <!-- Action to wipeout Cloud SQL data -->
   <servlet-mapping>
     <servlet-name>backend-servlet</servlet-name>

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

@@ -246,4 +246,14 @@
     <schedule>every 3 minutes</schedule>
     <target>backend</target>
   </cron>
+
+  <cron>
+    <url><![CDATA[/_dr/task/wipeOutContactHistoryPii]]></url>
+    <description>
+      This job runs weekly to wipe out PII fields of ContactHistory entities
+      that have been in the database for a certain period of time.
+    </description>
+    <schedule>every monday synchronized</schedule>
+    <target>backend</target>
+  </cron>
 </cronentries>

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

@@ -80,7 +80,7 @@ public class BackupDatastoreAction implements Runnable {
       logger.atInfo().log(message);
       response.setPayload(message);
     } catch (Throwable e) {
-      throw new InternalServerErrorException("Exception occurred while backing up datastore.", e);
+      throw new InternalServerErrorException("Exception occurred while backing up Datastore", e);
     }
   }
 }

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

@@ -118,10 +118,10 @@ public class BigqueryPollJobAction implements Runnable {
 
     // Check if the job ended with an error.
     if (job.getStatus().getErrorResult() != null) {
-      logger.atSevere().log("Bigquery job failed - %s - %s", jobRefString, job);
+      logger.atSevere().log("Bigquery job failed - %s - %s.", jobRefString, job);
       return false;
     }
-    logger.atInfo().log("Bigquery job succeeded - %s", jobRefString);
+    logger.atInfo().log("Bigquery job succeeded - %s.", jobRefString);
     return true;
   }
 

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

@@ -171,10 +171,10 @@ public class CheckBackupAction implements Runnable {
         ImmutableSet.copyOf(intersection(backup.getKinds(), kindsToLoad));
     String message = String.format("Datastore backup %s complete - ", backupName);
     if (exportedKindsToLoad.isEmpty()) {
-      message += "no kinds to load into BigQuery";
+      message += "no kinds to load into BigQuery.";
     } else {
       enqueueUploadBackupTask(backupId, backup.getExportFolderUrl(), exportedKindsToLoad);
-      message += "BigQuery load task enqueued";
+      message += "BigQuery load task enqueued.";
     }
     logger.atInfo().log(message);
     response.setPayload(message);

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

@@ -85,7 +85,7 @@ public class ExportDomainListsAction implements Runnable {
   @Override
   public void run() {
     ImmutableSet<String> realTlds = getTldsOfType(TldType.REAL);
-    logger.atInfo().log("Exporting domain lists for tlds %s", realTlds);
+    logger.atInfo().log("Exporting domain lists for TLDs %s.", realTlds);
     if (tm().isOfy()) {
       mrRunner
           .setJobName("Export domain lists")
@@ -145,7 +145,7 @@ public class ExportDomainListsAction implements Runnable {
       Registry registry = Registry.get(tld);
       if (registry.getDriveFolderId() == null) {
         logger.atInfo().log(
-            "Skipping registered domains export for TLD %s because Drive folder isn't specified",
+            "Skipping registered domains export for TLD %s because Drive folder isn't specified.",
             tld);
       } else {
         String resultMsg =

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

@@ -110,11 +110,11 @@ public class ExportPremiumTermsAction implements Runnable {
   private Optional<String> checkConfig(Registry registry) {
     if (isNullOrEmpty(registry.getDriveFolderId())) {
       logger.atInfo().log(
-          "Skipping premium terms export for TLD %s because Drive folder isn't specified", tld);
+          "Skipping premium terms export for TLD %s because Drive folder isn't specified.", tld);
       return Optional.of("Skipping export because no Drive folder is associated with this TLD");
     }
     if (!registry.getPremiumListName().isPresent()) {
-      logger.atInfo().log("No premium terms to export for TLD %s", tld);
+      logger.atInfo().log("No premium terms to export for TLD '%s'.", tld);
       return Optional.of("No premium lists configured");
     }
     return Optional.empty();

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

@@ -65,11 +65,11 @@ public class ExportReservedTermsAction implements Runnable {
       String resultMsg;
       if (registry.getReservedListNames().isEmpty() && isNullOrEmpty(registry.getDriveFolderId())) {
         resultMsg = "No reserved lists configured";
-        logger.atInfo().log("No reserved terms to export for TLD %s", tld);
+        logger.atInfo().log("No reserved terms to export for TLD '%s'.", tld);
       } else if (registry.getDriveFolderId() == null) {
         resultMsg = "Skipping export because no Drive folder is associated with this TLD";
         logger.atInfo().log(
-            "Skipping reserved terms export for TLD %s because Drive folder isn't specified", tld);
+            "Skipping reserved terms export for TLD %s because Drive folder isn't specified.", tld);
       } else {
         resultMsg = driveConnection.createOrUpdateFile(
             RESERVED_TERMS_FILENAME,

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

@@ -194,7 +194,7 @@ public final class SyncGroupMembersAction implements Runnable {
         }
       }
       logger.atInfo().log(
-          "Successfully synced contacts for registrar %s: added %d and removed %d",
+          "Successfully synced contacts for registrar %s: added %d and removed %d.",
           registrar.getRegistrarId(), totalAdded, totalRemoved);
     } catch (IOException e) {
       // Package up exception and re-throw with attached additional relevant info.

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

@@ -150,8 +150,7 @@ public class UpdateSnapshotViewAction implements Runnable {
       if (e.getDetails() != null && e.getDetails().getCode() == 404) {
         bigquery.tables().insert(ref.getProjectId(), ref.getDatasetId(), table).execute();
       } else {
-        logger.atWarning().withCause(e).log(
-            "UpdateSnapshotViewAction failed, caught exception %s", e.getDetails());
+        logger.atWarning().withCause(e).log("UpdateSnapshotViewAction errored out.");
       }
     }
   }

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

@@ -109,7 +109,7 @@ public class UploadDatastoreBackupAction implements Runnable {
       String message = uploadBackup(backupId, backupFolderUrl, Splitter.on(',').split(backupKinds));
       logger.atInfo().log("Loaded backup successfully: %s", message);
     } catch (Throwable e) {
-      logger.atSevere().withCause(e).log("Error loading backup");
+      logger.atSevere().withCause(e).log("Error loading backup.");
       if (e instanceof IllegalArgumentException) {
         throw new BadRequestException("Error calling load backup: " + e.getMessage(), e);
       } else {
@@ -148,12 +148,12 @@ public class UploadDatastoreBackupAction implements Runnable {
           getQueue(UpdateSnapshotViewAction.QUEUE));
 
       builder.append(String.format(" - %s:%s\n", projectId, jobId));
-      logger.atInfo().log("Submitted load job %s:%s", projectId, jobId);
+      logger.atInfo().log("Submitted load job %s:%s.", projectId, jobId);
     }
     return builder.toString();
   }
 
-  static String sanitizeForBigquery(String backupId) {
+  private static String sanitizeForBigquery(String backupId) {
     return backupId.replaceAll("[^a-zA-Z0-9_]", "_");
   }
 

core/src/main/java/google/registry/export/sheet/SheetSynchronizer.java

@@ -117,7 +117,7 @@ class SheetSynchronizer {
       BatchUpdateValuesResponse response =
           sheetsService.spreadsheets().values().batchUpdate(spreadsheetId, updateRequest).execute();
       Integer cellsUpdated = response.getTotalUpdatedCells();
-      logger.atInfo().log("Updated %d originalVals", cellsUpdated != null ? cellsUpdated : 0);
+      logger.atInfo().log("Updated %d originalVals.", cellsUpdated != null ? cellsUpdated : 0);
     }
 
     // Append extra rows if necessary
@@ -140,7 +140,7 @@ class SheetSynchronizer {
           .setInsertDataOption("INSERT_ROWS")
           .execute();
       logger.atInfo().log(
-          "Appended %d rows to range %s",
+          "Appended %d rows to range %s.",
           data.size() - originalVals.size(), appendResponse.getTableRange());
     // Clear the extra rows if necessary
     } else if (data.size() < originalVals.size()) {
@@ -155,7 +155,7 @@ class SheetSynchronizer {
                   new ClearValuesRequest())
               .execute();
       logger.atInfo().log(
-          "Cleared %d rows from range %s",
+          "Cleared %d rows from range %s.",
           originalVals.size() - data.size(), clearResponse.getClearedRange());
     }
   }

core/src/main/java/google/registry/flows/CheckApiAction.java

@@ -150,7 +150,7 @@ public class CheckApiAction implements Runnable {
       return fail(e.getResult().getMsg());
     } catch (Exception e) {
       metricBuilder.status(UNKNOWN_ERROR);
-      logger.atWarning().withCause(e).log("Unknown error");
+      logger.atWarning().withCause(e).log("Unknown error.");
       return fail("Invalid request");
     }
   }

core/src/main/java/google/registry/flows/EppController.java

@@ -130,12 +130,12 @@ public final class EppController {
     } catch (EppException | EppExceptionInProviderException e) {
       // The command failed. Send the client an error message, but only log at INFO since many of
       // these failures are innocuous or due to client error, so there's nothing we have to change.
-      logger.atInfo().withCause(e).log("Flow returned failure response");
+      logger.atInfo().withCause(e).log("Flow returned failure response.");
       EppException eppEx = (EppException) (e instanceof EppException ? e : e.getCause());
       return getErrorResponse(eppEx.getResult(), flowComponent.trid());
     } catch (Throwable e) {
       // Something bad and unexpected happened. Send the client a generic error, and log at SEVERE.
-      logger.atSevere().withCause(e).log("Unexpected failure in flow execution");
+      logger.atSevere().withCause(e).log("Unexpected failure in flow execution.");
       return getErrorResponse(Result.create(Code.COMMAND_FAILED), flowComponent.trid());
     }
   }

core/src/main/java/google/registry/flows/EppRequestHandler.java

@@ -84,7 +84,7 @@ public class EppRequestHandler {
         response.setHeader(ProxyHttpHeaders.LOGGED_IN, "true");
       }
     } catch (Exception e) {
-      logger.atWarning().withCause(e).log("handleEppCommand general exception");
+      logger.atWarning().withCause(e).log("handleEppCommand general exception.");
       response.setStatus(SC_BAD_REQUEST);
     }
   }

core/src/main/java/google/registry/flows/ExtensionManager.java

@@ -100,7 +100,7 @@ public final class ExtensionManager {
       throw new UndeclaredServiceExtensionException(undeclaredUrisThatError);
     }
     logger.atInfo().log(
-        "Client %s is attempting to run %s without declaring URIs %s on login",
+        "Client %s is attempting to run %s without declaring URIs %s on login.",
         registrarId, flowClass.getSimpleName(), undeclaredUris);
   }
 

core/src/main/java/google/registry/flows/certs/CertificateChecker.java

@@ -243,7 +243,7 @@ public class CertificateChecker {
      *    lastExpiringNotificationSentDate is greater than expirationWarningIntervalDays.
      */
     return !lastValidDate.after(now.plusDays(expirationWarningDays).toDate())
-        && (lastExpiringNotificationSentDate == START_OF_TIME
+        && (lastExpiringNotificationSentDate.equals(START_OF_TIME)
             || !lastExpiringNotificationSentDate
                 .plusDays(expirationWarningIntervalDays)
                 .toDate()

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. */
-  public final EppResponse runWithoutLogging() throws EppException {
+  private final 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/gcs/GcsUtils.java

@@ -141,7 +141,7 @@ public class GcsUtils implements Serializable {
       Blob blob = storage().get(blobId);
       return blob != null && blob.getSize() > 0;
     } catch (StorageException e) {
-      logger.atWarning().withCause(e).log("Failed to check if GCS file exists");
+      logger.atWarning().withCause(e).log("Failure while checking if GCS file exists.");
       return false;
     }
   }

core/src/main/java/google/registry/loadtest/LoadTestAction.java

@@ -256,7 +256,7 @@ public class LoadTestAction implements Runnable {
     }
     ImmutableList<TaskOptions> taskOptions = tasks.build();
     enqueue(taskOptions);
-    logger.atInfo().log("Added %d total load test tasks", taskOptions.size());
+    logger.atInfo().log("Added %d total load test tasks.", taskOptions.size());
   }
 
   private void validateAndLogRequest() {

core/src/main/java/google/registry/mapreduce/UnlockerOutput.java

@@ -58,7 +58,7 @@ public class UnlockerOutput<O> extends Output<O, Lock> {
 
   @Override
   public Lock finish(Collection<? extends OutputWriter<O>> writers) {
-    logger.atInfo().log("Mapreduce finished; releasing lock: %s", lock);
+    logger.atInfo().log("Mapreduce finished; releasing lock '%s'.", lock);
     lock.release();
     return lock;
   }

core/src/main/java/google/registry/mapreduce/inputs/EppResourceEntityReader.java

@@ -64,7 +64,7 @@ class EppResourceEntityReader<R extends EppResource> extends EppResourceBaseRead
       Key<? extends EppResource> key = nextQueryResult().getKey();
       EppResource resource = auditedOfy().load().key(key).now();
       if (resource == null) {
-        logger.atSevere().log("EppResourceIndex key %s points at a missing resource", key);
+        logger.atSevere().log("EppResourceIndex key %s points at a missing resource.", key);
         continue;
       }
       // Postfilter to distinguish polymorphic types (e.g. EppResources).

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

@@ -12,12 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package google.registry.backup;
+package google.registry.model;
 
 import com.google.apphosting.api.ApiProxy;
 import com.google.apphosting.api.ApiProxy.Environment;
 import com.google.common.collect.ImmutableMap;
-import java.io.Closeable;
+import google.registry.model.ofy.ObjectifyService;
+import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 
@@ -38,26 +39,39 @@ import java.lang.reflect.Proxy;
  * <p>Note that conversion from Objectify objects to Datastore {@code Entities} still requires the
  * Datastore service.
  */
-public class AppEngineEnvironment implements Closeable {
+public class AppEngineEnvironment {
 
-  private boolean isPlaceHolderNeeded;
+  private Environment environment;
 
   public AppEngineEnvironment() {
     this("PlaceholderAppId");
   }
 
   public AppEngineEnvironment(String appId) {
-    isPlaceHolderNeeded = ApiProxy.getCurrentEnvironment() == null;
-    // isPlaceHolderNeeded may be true when we are invoked in a test with AppEngineRule.
-    if (isPlaceHolderNeeded) {
-      ApiProxy.setEnvironmentForCurrentThread(createAppEngineEnvironment(appId));
-    }
+    environment = createAppEngineEnvironment(appId);
   }
 
-  @Override
-  public void close() {
-    if (isPlaceHolderNeeded) {
-      ApiProxy.setEnvironmentForCurrentThread(null);
+  public void setEnvironmentForCurrentThread() {
+    ApiProxy.setEnvironmentForCurrentThread(environment);
+    ObjectifyService.initOfy();
+  }
+
+  public void setEnvironmentForAllThreads() {
+    setEnvironmentForCurrentThread();
+    ApiProxy.setEnvironmentFactory(() -> environment);
+  }
+
+  public void unsetEnvironmentForCurrentThread() {
+    ApiProxy.clearEnvironmentForCurrentThread();
+  }
+
+  public void unsetEnvironmentForAllThreads() {
+    try {
+      Method method = ApiProxy.class.getDeclaredMethod("clearEnvironmentFactory");
+      method.setAccessible(true);
+      method.invoke(null);
+    } catch (NoSuchMethodException | InvocationTargetException | IllegalAccessException e) {
+      throw new RuntimeException(e);
     }
   }
 

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

@@ -18,13 +18,14 @@ import static com.google.common.base.Preconditions.checkState;
 
 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 java.util.concurrent.atomic.AtomicLong;
 
 /**
  * Allocates a globally unique {@link Long} number to use as an Ofy {@code @Id}.
  *
- * <p>In non-test environments the Id is generated by Datastore, whereas in tests it's from an
+ * <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.
  */
 public final class IdService {
@@ -35,13 +36,25 @@ public final class IdService {
    */
   private static final String APP_WIDE_ALLOCATION_KIND = "common";
 
-  /** Counts of used ids for use in unit tests. Outside tests this is never used. */
-  private static final AtomicLong nextTestId = new AtomicLong(1); // ids cannot be zero
+  /**
+   * Counts of used ids for use in unit tests or Beam.
+   *
+   * <p>Note that one should only use self-allocate Ids in Beam for entities whose Ids are not
+   * important and are not persisted back to the database, i. e. nowhere the uniqueness of the ID is
+   * required.
+   */
+  private static final AtomicLong nextSelfAllocatedId = new AtomicLong(1); // ids cannot be zero
+
+  private static final boolean isSelfAllocated() {
+    return RegistryEnvironment.UNITTEST.equals(RegistryEnvironment.get())
+        || "true".equals(System.getProperty(RegistryPipelineWorkerInitializer.PROPERTY, "false"));
+  }
 
   /** Allocates an id. */
+  // TODO(b/201547855): Find a way to allocate a unique ID without datastore.
   public static long allocateId() {
-    return RegistryEnvironment.UNITTEST.equals(RegistryEnvironment.get())
-        ? nextTestId.getAndIncrement()
+    return isSelfAllocated()
+        ? nextSelfAllocatedId.getAndIncrement()
         : DatastoreServiceFactory.getDatastoreService()
             .allocateIds(APP_WIDE_ALLOCATION_KIND, 1)
             .iterator()
@@ -49,13 +62,11 @@ public final class IdService {
             .getId();
   }
 
-  /** Resets the global test id counter (i.e. sets the next id to 1). */
+  /** Resets the global self-allocated id counter (i.e. sets the next id to 1). */
   @VisibleForTesting
-  public static void resetNextTestId() {
+  public static void resetSelfAllocatedId() {
     checkState(
-        RegistryEnvironment.UNITTEST.equals(RegistryEnvironment.get()),
-        "Can't call resetTestIdCounts() from RegistryEnvironment.%s",
-        RegistryEnvironment.get());
-    nextTestId.set(1); // ids cannot be zero
+        isSelfAllocated(), "Can only call resetSelfAllocatedId() in unit tests or Beam pipelines");
+    nextSelfAllocatedId.set(1); // ids cannot be zero
   }
 }

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

@@ -15,8 +15,12 @@
 package google.registry.model;
 
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
+import static org.joda.time.DateTimeZone.UTC;
 
+import com.google.common.flogger.FluentLogger;
+import com.google.common.flogger.StackSize;
 import com.googlecode.objectify.annotation.Ignore;
 import com.googlecode.objectify.annotation.OnLoad;
 import google.registry.model.translators.UpdateAutoTimestampTranslatorFactory;
@@ -40,6 +44,8 @@ import org.joda.time.DateTime;
 @Embeddable
 public class UpdateAutoTimestamp extends ImmutableObject {
 
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
   // When set to true, database converters/translators should do the auto update.  When set to
   // false, auto update should be suspended (this exists to allow us to preserve the original value
   // during a replay).
@@ -57,7 +63,16 @@ public class UpdateAutoTimestamp extends ImmutableObject {
   @PrePersist
   @PreUpdate
   void setTimestamp() {
-    if (autoUpdateEnabled() || lastUpdateTime == null) {
+    // On the off chance that this is called outside of a transaction, log it instead of failing
+    // with an exception from attempting to call jpaTm().getTransactionTime(), and then fall back
+    // to DateTime.now(UTC).
+    if (!jpaTm().inTransaction()) {
+      logger.atSevere().withStackTrace(StackSize.MEDIUM).log(
+          "Failed to update automatic timestamp because this wasn't called in a JPA transaction%s.",
+          ofyTm().inTransaction() ? " (but there is an open Ofy transaction)" : "");
+      timestamp = DateTime.now(UTC);
+      lastUpdateTime = DateTimeUtils.toZonedDateTime(timestamp);
+    } else if (autoUpdateEnabled() || lastUpdateTime == null) {
       timestamp = jpaTm().getTransactionTime();
       lastUpdateTime = DateTimeUtils.toZonedDateTime(timestamp);
     }

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

@@ -220,10 +220,10 @@ public class Cursor extends ImmutableObject implements DatastoreAndSqlEntity {
 
   private static String getScopeFromId(String id) {
     List<String> idSplit = Splitter.on('_').splitToList(id);
-    // The "parent" is always the crossTldKey; in order to find the scope (either Registry or
-    // cross-tld-key) we have to parse the part of the ID
+    // The key is always either the cross-tld-key or the key of a TLD (whose parent is the
+    // cross-tld-key).
     Key<?> scopeKey = Key.valueOf(idSplit.get(0));
-    return scopeKey.equals(getCrossTldKey()) ? GLOBAL : scopeKey.getName();
+    return scopeKey.getParent() == null ? GLOBAL : scopeKey.getName();
   }
 
   private static CursorType getTypeFromId(String id) {

core/src/main/java/google/registry/model/contact/ContactHistory.java

@@ -227,5 +227,11 @@ public class ContactHistory extends HistoryEntry implements SqlEntity {
       getInstance().parent = Key.create(ContactResource.class, contactRepoId);
       return this;
     }
+
+    public Builder wipeOutPii() {
+      getInstance().contactBase =
+          getInstance().getContactBase().get().asBuilder().wipeOut().build();
+      return this;
+    }
   }
 }

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

@@ -228,7 +228,7 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
           tm().transact(
                   () ->
                       jpaTm()
-                          .query(
+                          .criteriaQuery(
                               CriteriaQueryBuilder.create(clazz)
                                   .whereFieldIsIn(property, foreignKeys)
                                   .build())

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

@@ -32,6 +32,7 @@ import com.google.common.collect.Streams;
 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.InCrossTld;
 import google.registry.model.contact.ContactHistory;
 import google.registry.model.domain.DomainHistory;
@@ -128,7 +129,12 @@ public class DatastoreTransactionManager implements TransactionManager {
   }
 
   @Override
-  public void insertWithoutBackup(Object entity) {
+  public void insertAll(ImmutableObject... entities) {
+    putAll(entities);
+  }
+
+  @Override
+  public void insertWithoutBackup(ImmutableObject entity) {
     putWithoutBackup(entity);
   }
 
@@ -143,7 +149,7 @@ public class DatastoreTransactionManager implements TransactionManager {
   }
 
   @Override
-  public void putAll(Object... entities) {
+  public void putAll(ImmutableObject... entities) {
     syncIfTransactionless(
         getOfy().save().entities(toDatastoreEntities(ImmutableList.copyOf(entities))));
   }
@@ -154,7 +160,7 @@ public class DatastoreTransactionManager implements TransactionManager {
   }
 
   @Override
-  public void putWithoutBackup(Object entity) {
+  public void putWithoutBackup(ImmutableObject entity) {
     syncIfTransactionless(getOfy().saveWithoutBackup().entities(toDatastoreEntity(entity)));
   }
 
@@ -174,12 +180,12 @@ public class DatastoreTransactionManager implements TransactionManager {
   }
 
   @Override
-  public void updateAll(Object... entities) {
-    updateAll(ImmutableList.of(entities));
+  public void updateAll(ImmutableObject... entities) {
+    updateAll(ImmutableList.copyOf(entities));
   }
 
   @Override
-  public void updateWithoutBackup(Object entity) {
+  public void updateWithoutBackup(ImmutableObject entity) {
     putWithoutBackup(entity);
   }
 

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

@@ -283,7 +283,7 @@ public class Ofy {
         }
         sleeper.sleepUninterruptibly(Duration.millis(sleepMillis));
         logger.atInfo().withCause(e).log(
-            "Retrying %s, attempt %d", e.getClass().getSimpleName(), attempt);
+            "Retrying %s, attempt %d.", e.getClass().getSimpleName(), attempt);
       }
     }
   }

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

@@ -18,6 +18,10 @@ import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static google.registry.request.Action.Method.GET;
+import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
+import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
+import static javax.servlet.http.HttpServletResponse.SC_OK;
+import static org.joda.time.Duration.standardHours;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
@@ -26,15 +30,20 @@ import google.registry.model.UpdateAutoTimestamp;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
+import google.registry.model.server.Lock;
 import google.registry.persistence.transaction.Transaction;
 import google.registry.persistence.transaction.TransactionEntity;
 import google.registry.request.Action;
+import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.util.Clock;
+import google.registry.util.RequestStatusChecker;
 import java.io.IOException;
 import java.util.List;
+import java.util.Optional;
 import javax.inject.Inject;
 import javax.persistence.NoResultException;
+import org.joda.time.Duration;
 
 /** Cron task to replicate from Cloud SQL to datastore. */
 @Action(
@@ -55,11 +64,18 @@ public class ReplicateToDatastoreAction implements Runnable {
    */
   public static final int BATCH_SIZE = 200;
 
+  private static final Duration LEASE_LENGTH = standardHours(1);
+
   private final Clock clock;
+  private final RequestStatusChecker requestStatusChecker;
+  private final Response response;
 
   @Inject
-  public ReplicateToDatastoreAction(Clock clock) {
+  public ReplicateToDatastoreAction(
+      Clock clock, RequestStatusChecker requestStatusChecker, Response response) {
     this.clock = clock;
+    this.requestStatusChecker = requestStatusChecker;
+    this.response = response;
   }
 
   @VisibleForTesting
@@ -91,7 +107,7 @@ public class ReplicateToDatastoreAction implements Runnable {
    */
   @VisibleForTesting
   public void applyTransaction(TransactionEntity txnEntity) {
-    logger.atInfo().log("Applying a single transaction Cloud SQL -> Cloud Datastore");
+    logger.atInfo().log("Applying a single transaction Cloud SQL -> Cloud Datastore.");
     try (UpdateAutoTimestamp.DisableAutoUpdateResource disabler =
         UpdateAutoTimestamp.disableAutoUpdate()) {
       ofyTm()
@@ -120,21 +136,21 @@ public class ReplicateToDatastoreAction implements Runnable {
                 }
 
                 logger.atInfo().log(
-                    "Applying transaction %s to Cloud Datastore", txnEntity.getId());
+                    "Applying transaction %s to Cloud Datastore.", txnEntity.getId());
 
                 // At this point, we know txnEntity is the correct next transaction, so write it
-                // to datastore.
+                // to Datastore.
                 try {
                   Transaction.deserialize(txnEntity.getContents()).writeToDatastore();
                 } catch (IOException e) {
-                  throw new RuntimeException("Error during transaction deserialization.", e);
+                  throw new RuntimeException("Error during transaction deserialization", e);
                 }
 
-                // Write the updated last transaction id to datastore as part of this datastore
+                // Write the updated last transaction id to Datastore as part of this Datastore
                 // transaction.
                 auditedOfy().save().entity(lastSqlTxn.cloneWithNewTransactionId(nextTxnId));
                 logger.atInfo().log(
-                    "Finished applying single transaction Cloud SQL -> Cloud Datastore");
+                    "Finished applying single transaction Cloud SQL -> Cloud Datastore.");
               });
     }
   }
@@ -143,24 +159,55 @@ public class ReplicateToDatastoreAction implements Runnable {
   public void run() {
     MigrationState state = DatabaseMigrationStateSchedule.getValueAtTime(clock.nowUtc());
     if (!state.getReplayDirection().equals(ReplayDirection.SQL_TO_DATASTORE)) {
-      logger.atInfo().log(
-          "Skipping ReplicateToDatastoreAction because we are in migration phase %s.", state);
+      String message =
+          String.format(
+              "Skipping ReplicateToDatastoreAction because we are in migration phase %s.", state);
+      logger.atInfo().log(message);
+      // App Engine will retry on any non-2xx status code, which we don't want in this case.
+      response.setStatus(SC_NO_CONTENT);
+      response.setPayload(message);
       return;
     }
-    // TODO(b/181758163): Deal with objects that don't exist in Cloud SQL, e.g. ForeignKeyIndex,
-    // EppResourceIndex.
-    logger.atInfo().log("Processing transaction replay batch Cloud SQL -> Cloud Datastore");
-    int numTransactionsReplayed = 0;
-    for (TransactionEntity txnEntity : getTransactionBatch()) {
-      try {
-        applyTransaction(txnEntity);
-      } catch (Throwable t) {
-        logger.atSevere().withCause(t).log("Errored out replaying files");
-        return;
-      }
-      numTransactionsReplayed++;
+    Optional<Lock> lock =
+        Lock.acquire(
+            this.getClass().getSimpleName(), null, LEASE_LENGTH, requestStatusChecker, false);
+    if (!lock.isPresent()) {
+      String message = "Can't acquire ReplicateToDatastoreAction lock, aborting.";
+      logger.atSevere().log(message);
+      // App Engine will retry on any non-2xx status code, which we don't want in this case.
+      response.setStatus(SC_NO_CONTENT);
+      response.setPayload(message);
+      return;
     }
-    logger.atInfo().log(
-        "Replayed %d transactions from Cloud SQL -> Datastore", numTransactionsReplayed);
+    try {
+      logger.atInfo().log("Processing transaction replay batch Cloud SQL -> Cloud Datastore.");
+      int numTransactionsReplayed = replayAllTransactions();
+      String resultMessage =
+          String.format(
+              "Replayed %d transaction(s) from Cloud SQL -> Datastore.", numTransactionsReplayed);
+      logger.atInfo().log(resultMessage);
+      response.setPayload(resultMessage);
+      response.setStatus(SC_OK);
+    } catch (Throwable t) {
+      String message = "Errored out replaying files.";
+      logger.atSevere().withCause(t).log(message);
+      response.setStatus(SC_INTERNAL_SERVER_ERROR);
+      response.setPayload(message);
+    } finally {
+      lock.ifPresent(Lock::release);
+    }
+  }
+
+  private int replayAllTransactions() {
+    int numTransactionsReplayed = 0;
+    List<TransactionEntity> transactionBatch;
+    do {
+      transactionBatch = getTransactionBatch();
+      for (TransactionEntity transaction : transactionBatch) {
+        applyTransaction(transaction);
+        numTransactionsReplayed++;
+      }
+    } while (!transactionBatch.isEmpty());
+    return numTransactionsReplayed;
   }
 }

core/src/main/java/google/registry/model/reporting/HistoryEntryDao.java

@@ -24,6 +24,7 @@ import static google.registry.util.DateTimeUtils.START_OF_TIME;
 
 import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Streams;
 import google.registry.model.EppResource;
 import google.registry.model.contact.ContactHistory;
@@ -49,6 +50,25 @@ import org.joda.time.DateTime;
  */
 public class HistoryEntryDao {
 
+  public static ImmutableMap<Class<? extends EppResource>, Class<? extends HistoryEntry>>
+      RESOURCE_TYPES_TO_HISTORY_TYPES =
+          ImmutableMap.of(
+              ContactResource.class,
+              ContactHistory.class,
+              DomainBase.class,
+              DomainHistory.class,
+              HostResource.class,
+              HostHistory.class);
+
+  public static ImmutableMap<Class<? extends HistoryEntry>, String> REPO_ID_FIELD_NAMES =
+      ImmutableMap.of(
+          ContactHistory.class,
+          "contactRepoId",
+          DomainHistory.class,
+          "domainRepoId",
+          HostHistory.class,
+          "hostRepoId");
+
   /** Loads all history objects in the times specified, including all types. */
   public static ImmutableList<HistoryEntry> loadAllHistoryObjects(
       DateTime afterTime, DateTime beforeTime) {
@@ -164,7 +184,7 @@ public class HistoryEntryDao {
   private static <T extends HistoryEntry> Stream<T> loadHistoryObjectFromSqlByRegistrars(
       Class<T> historyClass, ImmutableCollection<String> registrarIds) {
     return jpaTm()
-        .query(
+        .criteriaQuery(
             CriteriaQueryBuilder.create(historyClass)
                 .whereFieldIsIn("clientId", registrarIds)
                 .build())
@@ -188,34 +208,32 @@ public class HistoryEntryDao {
 
     return ImmutableList.sortedCopyOf(
         Comparator.comparing(HistoryEntry::getModificationTime),
-        jpaTm().query(criteriaQuery).getResultList());
+        jpaTm().criteriaQuery(criteriaQuery).getResultList());
   }
 
   private static Class<? extends HistoryEntry> getHistoryClassFromParent(
       Class<? extends EppResource> parent) {
-    if (parent.equals(ContactResource.class)) {
-      return ContactHistory.class;
-    } else if (parent.equals(DomainBase.class)) {
-      return DomainHistory.class;
-    } else if (parent.equals(HostResource.class)) {
-      return HostHistory.class;
+    if (!RESOURCE_TYPES_TO_HISTORY_TYPES.containsKey(parent)) {
+      throw new IllegalArgumentException(
+          String.format("Unknown history type for parent %s", parent.getName()));
     }
-    throw new IllegalArgumentException(
-        String.format("Unknown history type for parent %s", parent.getName()));
+    return RESOURCE_TYPES_TO_HISTORY_TYPES.get(parent);
   }
 
   private static String getRepoIdFieldNameFromHistoryClass(
       Class<? extends HistoryEntry> historyClass) {
-    return historyClass.equals(ContactHistory.class)
-        ? "contactRepoId"
-        : historyClass.equals(DomainHistory.class) ? "domainRepoId" : "hostRepoId";
+    if (!REPO_ID_FIELD_NAMES.containsKey(historyClass)) {
+      throw new IllegalArgumentException(
+          String.format("Unknown history type %s", historyClass.getName()));
+    }
+    return REPO_ID_FIELD_NAMES.get(historyClass);
   }
 
   private static <T extends HistoryEntry> List<T> loadAllHistoryObjectsFromSql(
       Class<T> historyClass, DateTime afterTime, DateTime beforeTime) {
     CriteriaBuilder criteriaBuilder = jpaTm().getEntityManager().getCriteriaBuilder();
     return jpaTm()
-        .query(
+        .criteriaQuery(
             CriteriaQueryBuilder.create(historyClass)
                 .where("modificationTime", criteriaBuilder::greaterThanOrEqualTo, afterTime)
                 .where("modificationTime", criteriaBuilder::lessThanOrEqualTo, beforeTime)

core/src/main/java/google/registry/model/server/Lock.java

@@ -15,6 +15,7 @@
 package google.registry.model.server;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.DateTimeUtils.isAtOrAfter;
 import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
@@ -35,6 +36,7 @@ import google.registry.util.RequestStatusChecker;
 import google.registry.util.RequestStatusCheckerImpl;
 import java.io.Serializable;
 import java.util.Optional;
+import java.util.function.Supplier;
 import javax.annotation.Nullable;
 import javax.persistence.Column;
 import javax.persistence.IdClass;
@@ -215,45 +217,45 @@ public class Lock extends ImmutableObject implements DatastoreAndSqlEntity, Seri
     // It's important to use transactNew rather than transact, because a Lock can be used to control
     // access to resources like GCS that can't be transactionally rolled back. Therefore, the lock
     // must be definitively acquired before it is used, even when called inside another transaction.
+    Supplier<AcquireResult> lockAcquirer =
+        () -> {
+          DateTime now = tm().getTransactionTime();
+
+          // Checking if an unexpired lock still exists - if so, the lock can't be acquired.
+          Lock lock =
+              tm().loadByKeyIfPresent(
+                      VKey.create(
+                          Lock.class,
+                          new LockId(resourceName, scope),
+                          Key.create(Lock.class, lockId)))
+                  .orElse(null);
+          if (lock != null) {
+            logger.atInfo().log(
+                "Loaded existing lock: %s for request: %s", lock.lockId, lock.requestLogId);
+          }
+          LockState lockState;
+          if (lock == null) {
+            lockState = LockState.FREE;
+          } else if (isAtOrAfter(now, lock.expirationTime)) {
+            lockState = LockState.TIMED_OUT;
+          } else if (checkThreadRunning && !requestStatusChecker.isRunning(lock.requestLogId)) {
+            lockState = LockState.OWNER_DIED;
+          } else {
+            lockState = LockState.IN_USE;
+            return AcquireResult.create(now, lock, null, lockState);
+          }
+
+          Lock newLock =
+              create(resourceName, scope, requestStatusChecker.getLogId(), now, leaseLength);
+          // Locks are not parented under an EntityGroupRoot (so as to avoid write
+          // contention) and don't need to be backed up.
+          tm().putIgnoringReadOnly(newLock);
+
+          return AcquireResult.create(now, lock, newLock, lockState);
+        };
+    // In ofy, backup is determined per-action, but in SQL it's determined per-transaction
     AcquireResult acquireResult =
-        tm().transactNew(
-                () -> {
-                  DateTime now = tm().getTransactionTime();
-
-                  // Checking if an unexpired lock still exists - if so, the lock can't be acquired.
-                  Lock lock =
-                      tm().loadByKeyIfPresent(
-                              VKey.create(
-                                  Lock.class,
-                                  new LockId(resourceName, scope),
-                                  Key.create(Lock.class, lockId)))
-                          .orElse(null);
-                  if (lock != null) {
-                    logger.atInfo().log(
-                        "Loaded existing lock: %s for request: %s", lock.lockId, lock.requestLogId);
-                  }
-                  LockState lockState;
-                  if (lock == null) {
-                    lockState = LockState.FREE;
-                  } else if (isAtOrAfter(now, lock.expirationTime)) {
-                    lockState = LockState.TIMED_OUT;
-                  } else if (checkThreadRunning
-                      && !requestStatusChecker.isRunning(lock.requestLogId)) {
-                    lockState = LockState.OWNER_DIED;
-                  } else {
-                    lockState = LockState.IN_USE;
-                    return AcquireResult.create(now, lock, null, lockState);
-                  }
-
-                  Lock newLock =
-                      create(
-                          resourceName, scope, requestStatusChecker.getLogId(), now, leaseLength);
-                  // Locks are not parented under an EntityGroupRoot (so as to avoid write
-                  // contention) and don't need to be backed up.
-                  tm().putIgnoringReadOnly(newLock);
-
-                  return AcquireResult.create(now, lock, newLock, lockState);
-                });
+        tm().isOfy() ? tm().transactNew(lockAcquirer) : jpaTm().transactWithoutBackup(lockAcquirer);
 
     logAcquireResult(acquireResult);
     lockMetrics.recordAcquire(resourceName, scope, acquireResult.lockState());
@@ -263,34 +265,41 @@ public class Lock extends ImmutableObject implements DatastoreAndSqlEntity, Seri
   /** Release the lock. */
   public void release() {
     // Just use the default clock because we aren't actually doing anything that will use the clock.
-    tm().transact(
-            () -> {
-              // To release a lock, check that no one else has already obtained it and if not
-              // delete it. If the lock in Datastore was different then this lock is gone already;
-              // this can happen if release() is called around the expiration time and the lock
-              // expires underneath us.
-              VKey<Lock> key =
-                  VKey.create(
-                      Lock.class, new LockId(resourceName, tld), Key.create(Lock.class, lockId));
-              Lock loadedLock = tm().loadByKeyIfPresent(key).orElse(null);
-              if (Lock.this.equals(loadedLock)) {
-                // Use deleteWithoutBackup() so that we don't create a commit log entry for deleting
-                // the lock.
-                logger.atInfo().log("Deleting lock: %s", lockId);
-                tm().deleteIgnoringReadOnly(key);
+    Supplier<Void> lockReleaser =
+        () -> {
+          // To release a lock, check that no one else has already obtained it and if not
+          // delete it. If the lock in Datastore was different then this lock is gone already;
+          // this can happen if release() is called around the expiration time and the lock
+          // expires underneath us.
+          VKey<Lock> key =
+              VKey.create(
+                  Lock.class, new LockId(resourceName, tld), Key.create(Lock.class, lockId));
+          Lock loadedLock = tm().loadByKeyIfPresent(key).orElse(null);
+          if (Lock.this.equals(loadedLock)) {
+            // Use deleteIgnoringReadOnly() so that we don't create a commit log entry for deleting
+            // the lock.
+            logger.atInfo().log("Deleting lock: %s", lockId);
+            tm().deleteIgnoringReadOnly(key);
 
-                lockMetrics.recordRelease(
-                    resourceName, tld, new Duration(acquiredTime, tm().getTransactionTime()));
-              } else {
-                logger.atSevere().log(
-                    "The lock we acquired was transferred to someone else before we"
-                        + " released it! Did action take longer than lease length?"
-                        + " Our lock: %s, current lock: %s",
-                    Lock.this, loadedLock);
-                logger.atInfo().log(
-                    "Not deleting lock: %s - someone else has it: %s", lockId, loadedLock);
-              }
-            });
+            lockMetrics.recordRelease(
+                resourceName, tld, new Duration(acquiredTime, tm().getTransactionTime()));
+          } else {
+            logger.atSevere().log(
+                "The lock we acquired was transferred to someone else before we"
+                    + " released it! Did action take longer than lease length?"
+                    + " Our lock: %s, current lock: %s",
+                Lock.this, loadedLock);
+            logger.atInfo().log(
+                "Not deleting lock: %s - someone else has it: %s", lockId, loadedLock);
+          }
+          return null;
+        };
+    // In ofy, backup is determined per-action, but in SQL it's determined per-transaction
+    if (tm().isOfy()) {
+      tm().transact(lockReleaser);
+    } else {
+      jpaTm().transactWithoutBackup(lockReleaser);
+    }
   }
 
   static class LockId extends ImmutableObject implements Serializable {

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

@@ -30,10 +30,10 @@ public class ReservedListDao {
   /** Persist a new reserved list to Cloud SQL. */
   public static void save(ReservedList reservedList) {
     checkArgumentNotNull(reservedList, "Must specify reservedList");
-    logger.atInfo().log("Saving reserved list %s to Cloud SQL", reservedList.getName());
+    logger.atInfo().log("Saving reserved list %s to Cloud SQL.", reservedList.getName());
     jpaTm().transact(() -> jpaTm().insert(reservedList));
     logger.atInfo().log(
-        "Saved reserved list %s with %d entries to Cloud SQL",
+        "Saved reserved list %s with %d entries to Cloud SQL.",
         reservedList.getName(), reservedList.getReservedListEntries().size());
   }
 

core/src/main/java/google/registry/module/ServletBase.java

@@ -51,7 +51,7 @@ public class ServletBase extends HttpServlet {
     // registered if metric reporter starts up correctly.
     try {
       metricReporter.get().startAsync().awaitRunning(java.time.Duration.ofSeconds(10));
-      logger.atInfo().log("Started up MetricReporter");
+      logger.atInfo().log("Started up MetricReporter.");
       LifecycleManager.getInstance()
           .setShutdownHook(
               () -> {
@@ -60,7 +60,7 @@ public class ServletBase extends HttpServlet {
                       .get()
                       .stopAsync()
                       .awaitTerminated(java.time.Duration.ofSeconds(10));
-                  logger.atInfo().log("Shut down MetricReporter");
+                  logger.atInfo().log("Shut down MetricReporter.");
                 } catch (TimeoutException e) {
                   logger.atSevere().withCause(e).log("Failed to stop MetricReporter.");
                 }
@@ -72,13 +72,13 @@ public class ServletBase extends HttpServlet {
 
   @Override
   public void service(HttpServletRequest req, HttpServletResponse rsp) throws IOException {
-    logger.atInfo().log("Received %s request", getClass().getSimpleName());
+    logger.atInfo().log("Received %s request.", getClass().getSimpleName());
     DateTime startTime = clock.nowUtc();
     try {
       requestHandler.handleRequest(req, rsp);
     } finally {
       logger.atInfo().log(
-          "Finished %s request. Latency: %.3fs",
+          "Finished %s request. Latency: %.3fs.",
           getClass().getSimpleName(), (clock.nowUtc().getMillis() - startTime.getMillis()) / 1000d);
     }
   }

core/src/main/java/google/registry/module/backend/BackendModule.java

@@ -1,59 +0,0 @@
-// Copyright 2017 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.module.backend;
-
-import static google.registry.model.tld.Registries.assertTldExists;
-import static google.registry.model.tld.Registries.assertTldsExist;
-import static google.registry.request.RequestParameters.extractOptionalDatetimeParameter;
-import static google.registry.request.RequestParameters.extractRequiredParameter;
-import static google.registry.request.RequestParameters.extractSetOfParameters;
-
-import com.google.common.collect.ImmutableSet;
-import dagger.Module;
-import dagger.Provides;
-import google.registry.batch.ExpandRecurringBillingEventsAction;
-import google.registry.request.Parameter;
-import google.registry.request.RequestParameters;
-import java.util.Optional;
-import javax.servlet.http.HttpServletRequest;
-import org.joda.time.DateTime;
-
-/**
- * Dagger module for injecting common settings for all Backend tasks.
- */
-@Module
-public class BackendModule {
-
-  @Provides
-  @Parameter(RequestParameters.PARAM_TLD)
-  static String provideTld(HttpServletRequest req) {
-    return assertTldExists(extractRequiredParameter(req, RequestParameters.PARAM_TLD));
-  }
-
-  @Provides
-  @Parameter(RequestParameters.PARAM_TLDS)
-  static ImmutableSet<String> provideTlds(HttpServletRequest req) {
-    ImmutableSet<String> tlds = extractSetOfParameters(req, RequestParameters.PARAM_TLDS);
-    assertTldsExist(tlds);
-    return tlds;
-  }
-
-  @Provides
-  @Parameter("cursorTime")
-  static Optional<DateTime> provideCursorTime(HttpServletRequest req) {
-    return extractOptionalDatetimeParameter(
-        req, ExpandRecurringBillingEventsAction.PARAM_CURSOR_TIME);
-  }
-}

core/src/main/java/google/registry/module/backend/BackendRequestComponent.java

@@ -33,6 +33,7 @@ import google.registry.batch.ResaveAllEppResourcesAction;
 import google.registry.batch.ResaveEntityAction;
 import google.registry.batch.SendExpiringCertificateNotificationEmailAction;
 import google.registry.batch.WipeOutCloudSqlAction;
+import google.registry.batch.WipeOutContactHistoryPiiAction;
 import google.registry.batch.WipeoutDatastoreAction;
 import google.registry.cron.CommitLogFanoutAction;
 import google.registry.cron.CronModule;
@@ -94,7 +95,6 @@ import google.registry.tools.javascrap.CreateSyntheticHistoryEntriesAction;
 @RequestScope
 @Subcomponent(
     modules = {
-      BackendModule.class,
       BackupModule.class,
       BatchModule.class,
       BillingModule.class,
@@ -220,6 +220,8 @@ interface BackendRequestComponent {
 
   WipeoutDatastoreAction wipeoutDatastoreAction();
 
+  WipeOutContactHistoryPiiAction wipeOutContactHistoryPiiAction();
+
   @Subcomponent.Builder
   abstract class Builder implements RequestComponentBuilder<BackendRequestComponent> {
 

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

@@ -303,7 +303,7 @@ public abstract class PersistenceModule {
       overrides.put(Environment.PASS, credential.password());
     } catch (Throwable e) {
       // TODO(b/184631990): after SQL becomes primary, throw an exception to fail fast
-      logger.atSevere().withCause(e).log("Failed to get SQL credential from Secret Manager");
+      logger.atSevere().withCause(e).log("Failed to get SQL credential from Secret Manager.");
     }
   }
 

core/src/main/java/google/registry/persistence/transaction/DummyJpaTransactionManager.java

@@ -20,11 +20,11 @@ import java.lang.reflect.Proxy;
  * A dummy implementation for {@link JpaTransactionManager} which throws exception when any of its
  * method is invoked.
  *
- * <p>This is used to initialize the {@link TransactionManagerFactory#jpaTm} when running unit
+ * <p>This is used to initialize the {@link TransactionManagerFactory#jpaTm()} when running unit
  * tests, because obviously we cannot connect to the actual Cloud SQL backend in a unit test.
  *
- * <p>If a unit test needs to access the Cloud SQL database, it must add JpaTransactionManagerRule
- * as a JUnit rule in the test class.
+ * <p>If a unit test needs to access the Cloud SQL database, it must add {@code
+ * JpaTransactionManagerExtension} as a JUnit extension in the test class.
  */
 public class DummyJpaTransactionManager {
 
@@ -37,7 +37,7 @@ public class DummyJpaTransactionManager {
             (proxy, method, args) -> {
               throw new UnsupportedOperationException(
                   "JpaTransactionManager was not initialized as the runtime is detected as"
-                      + " Unittest. Add JpaTransactionManagerRule in the unit test for"
+                      + " Unittest. Add JpaTransactionManagerExtension in the unit test for"
                       + " initialization.");
             });
   }

core/src/main/java/google/registry/persistence/transaction/JpaTransactionManager.java

@@ -36,7 +36,7 @@ public interface JpaTransactionManager extends TransactionManager {
   <T> TypedQuery<T> query(String sqlString, Class<T> resultClass);
 
   /** Creates a JPA SQU query for the given criteria query. */
-  <T> TypedQuery<T> query(CriteriaQuery<T> criteriaQuery);
+  <T> TypedQuery<T> criteriaQuery(CriteriaQuery<T> criteriaQuery);
 
   /**
    * Creates a JPA SQL query for the given query string.

core/src/main/java/google/registry/persistence/transaction/JpaTransactionManagerImpl.java

@@ -135,7 +135,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public <T> TypedQuery<T> query(CriteriaQuery<T> criteriaQuery) {
+  public <T> TypedQuery<T> criteriaQuery(CriteriaQuery<T> criteriaQuery) {
     return new DetachingTypedQuery<>(getEntityManager().createQuery(criteriaQuery));
   }
 
@@ -186,10 +186,10 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
             // Error is unchecked!
             try {
               txn.rollback();
-              logger.atWarning().log("Error during transaction; transaction rolled back");
+              logger.atWarning().log("Error during transaction; transaction rolled back.");
             } catch (Throwable rollbackException) {
               logger.atSevere().withCause(rollbackException).log(
-                  "Rollback failed; suppressing error");
+                  "Rollback failed; suppressing error.");
             }
             throw e;
           } finally {
@@ -218,9 +218,9 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
       // Error is unchecked!
       try {
         txn.rollback();
-        logger.atWarning().log("Error during transaction; transaction rolled back");
+        logger.atWarning().log("Error during transaction; transaction rolled back.");
       } catch (Throwable rollbackException) {
-        logger.atSevere().withCause(rollbackException).log("Rollback failed; suppressing error");
+        logger.atSevere().withCause(rollbackException).log("Rollback failed; suppressing error.");
       }
       throw e;
     } finally {
@@ -313,7 +313,12 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public void insertWithoutBackup(Object entity) {
+  public void insertAll(ImmutableObject... entities) {
+    insertAll(ImmutableSet.copyOf(entities));
+  }
+
+  @Override
+  public void insertWithoutBackup(ImmutableObject entity) {
     insert(entity);
   }
 
@@ -335,7 +340,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public void putAll(Object... entities) {
+  public void putAll(ImmutableObject... entities) {
     checkArgumentNotNull(entities, "entities must be specified");
     assertInTransaction();
     for (Object entity : entities) {
@@ -351,7 +356,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public void putWithoutBackup(Object entity) {
+  public void putWithoutBackup(ImmutableObject entity) {
     put(entity);
   }
 
@@ -381,12 +386,12 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public void updateAll(Object... entities) {
-    updateAll(ImmutableList.of(entities));
+  public void updateAll(ImmutableObject... entities) {
+    updateAll(ImmutableList.copyOf(entities));
   }
 
   @Override
-  public void updateWithoutBackup(Object entity) {
+  public void updateWithoutBackup(ImmutableObject entity) {
     update(entity);
   }
 

core/src/main/java/google/registry/persistence/transaction/ReadOnlyCheckingQuery.java

@@ -21,6 +21,7 @@ import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Stream;
 import javax.persistence.FlushModeType;
 import javax.persistence.LockModeType;
 import javax.persistence.Parameter;
@@ -41,6 +42,11 @@ class ReadOnlyCheckingQuery implements Query {
     return delegate.getResultList();
   }
 
+  @Override
+  public Stream getResultStream() {
+    return delegate.getResultStream();
+  }
+
   @Override
   public Object getSingleResult() {
     return delegate.getSingleResult();

core/src/main/java/google/registry/persistence/transaction/Transaction.java

@@ -242,7 +242,7 @@ public class Transaction extends ImmutableObject implements Buildable {
       if (entity instanceof DatastoreEntity) {
         ((DatastoreEntity) entity).beforeDatastoreSaveOnReplay();
       }
-      ofyTm().put(entity);
+      ofyTm().putIgnoringReadOnly(entity);
     }
 
     @Override
@@ -280,7 +280,7 @@ public class Transaction extends ImmutableObject implements Buildable {
 
     @Override
     public void writeToDatastore() {
-      ofyTm().delete(key);
+      ofyTm().deleteIgnoringReadOnly(key);
     }
 
     @Override

core/src/main/java/google/registry/persistence/transaction/TransactionManager.java

@@ -17,6 +17,7 @@ package google.registry.persistence.transaction;
 import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import google.registry.model.ImmutableObject;
 import google.registry.model.annotations.InCrossTld;
 import google.registry.persistence.VKey;
 import java.util.NoSuchElementException;
@@ -97,6 +98,9 @@ public interface TransactionManager {
   /** Persists all new entities in the database, throws exception if any entity already exists. */
   void insertAll(ImmutableCollection<?> entities);
 
+  /** Persists all new entities in the database, throws exception if any entity already exists. */
+  void insertAll(ImmutableObject... entities);
+
   /**
    * Persists a new entity in the database without writing any backup if the underlying database is
    * Datastore.
@@ -107,7 +111,7 @@ public interface TransactionManager {
    * "WithoutBackup" in its method name because it is not necessary to have the backup mechanism in
    * SQL.
    */
-  void insertWithoutBackup(Object entity);
+  void insertWithoutBackup(ImmutableObject entity);
 
   /**
    * Persists all new entities in the database without writing any backup if the underlying database
@@ -125,7 +129,7 @@ public interface TransactionManager {
   void put(Object entity);
 
   /** Persists all new entities or updates the existing entities in the database. */
-  void putAll(Object... entities);
+  void putAll(ImmutableObject... entities);
 
   /** Persists all new entities or updates the existing entities in the database. */
   void putAll(ImmutableCollection<?> entities);
@@ -140,7 +144,7 @@ public interface TransactionManager {
    * "WithoutBackup" in its method name because it is not necessary to have the backup mechanism in
    * SQL.
    */
-  void putWithoutBackup(Object entity);
+  void putWithoutBackup(ImmutableObject entity);
 
   /**
    * Persists all new entities or update the existing entities in the database without writing
@@ -161,7 +165,7 @@ public interface TransactionManager {
   void updateAll(ImmutableCollection<?> entities);
 
   /** Updates all entities in the database, throws exception if any entity does not exist. */
-  void updateAll(Object... entities);
+  void updateAll(ImmutableObject... entities);
 
   /**
    * Updates an entity in the database without writing commit logs if the underlying database is
@@ -173,7 +177,7 @@ public interface TransactionManager {
    * "WithoutBackup" in its method name because it is not necessary to have the backup mechanism in
    * SQL.
    */
-  void updateWithoutBackup(Object entity);
+  void updateWithoutBackup(ImmutableObject entity);
 
   /**
    * Updates all entities in the database without writing any backup if the underlying database is

core/src/main/java/google/registry/persistence/transaction/TransactionManagerFactory.java

@@ -132,7 +132,9 @@ public class TransactionManagerFactory {
   /**
    * Sets the return of {@link #tm()} to the given instance of {@link TransactionManager}.
    *
-   * <p>Used when overriding the per-test transaction manager for dual-database tests.
+   * <p>Used when overriding the per-test transaction manager for dual-database tests. Should be
+   * matched with a corresponding invocation of {@link #removeTmOverrideForTest()} either at the end
+   * of the test or in an <code>@AfterEach</code> handler.
    */
   @VisibleForTesting
   public static void setTmForTest(TransactionManager newTm) {

core/src/main/java/google/registry/rdap/RdapActionBase.java

@@ -139,7 +139,7 @@ public abstract class RdapActionBase implements Runnable {
           pathProper.startsWith(getActionPath()),
           "%s doesn't start with %s", pathProper, getActionPath());
       String pathSearchString = pathProper.substring(getActionPath().length());
-      logger.atInfo().log("path search string: '%s'", pathSearchString);
+      logger.atInfo().log("path search string: '%s'.", pathSearchString);
 
       ReplyPayloadBase replyObject =
           getJsonObjectForResource(pathSearchString, requestMethod == Action.Method.HEAD);
@@ -153,14 +153,14 @@ public abstract class RdapActionBase implements Runnable {
       setPayload(replyObject);
       metricInformationBuilder.setStatusCode(SC_OK);
     } catch (HttpException e) {
-      logger.atInfo().withCause(e).log("Error in RDAP");
+      logger.atInfo().withCause(e).log("Error in RDAP.");
       setError(e.getResponseCode(), e.getResponseCodeString(), e.getMessage());
     } catch (URISyntaxException | IllegalArgumentException e) {
-      logger.atInfo().withCause(e).log("Bad request in RDAP");
+      logger.atInfo().withCause(e).log("Bad request in RDAP.");
       setError(SC_BAD_REQUEST, "Bad Request", "Not a valid " + getHumanReadableObjectTypeName());
     } catch (RuntimeException e) {
       setError(SC_INTERNAL_SERVER_ERROR, "Internal Server Error", "An error was encountered");
-      logger.atSevere().withCause(e).log("Exception encountered while processing RDAP command");
+      logger.atSevere().withCause(e).log("Exception encountered while processing RDAP command.");
     }
     rdapMetrics.updateMetrics(metricInformationBuilder.build());
   }

core/src/main/java/google/registry/rdap/RdapDomainSearchAction.java

@@ -448,7 +448,7 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
           if (hostKey != null) {
             builder.add(hostKey);
           } else {
-            logger.atWarning().log("Host key unexpectedly null");
+            logger.atWarning().log("Host key unexpectedly null.");
           }
         }
       }
@@ -591,7 +591,7 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
                               cursorString.get());
                     }
                     jpaTm()
-                        .query(queryBuilder.build())
+                        .criteriaQuery(queryBuilder.build())
                         .getResultStream()
                         .filter(this::isAuthorized)
                         .forEach(

core/src/main/java/google/registry/rdap/RdapJsonFormatter.java

@@ -332,7 +332,7 @@ public class RdapJsonFormatter {
     builder.statusBuilder().addAll(status);
     if (status.isEmpty()) {
       logger.atWarning().log(
-          "Domain %s (ROID %s) doesn't have any status",
+          "Domain %s (ROID %s) doesn't have any status.",
           domainBase.getDomainName(), domainBase.getRepoId());
     }
     // RDAP Response Profile 2.6.3, must have a notice about statuses. That is in {@link
@@ -741,7 +741,7 @@ public class RdapJsonFormatter {
       if (registrarContacts.stream()
           .noneMatch(contact -> contact.roles().contains(RdapEntity.Role.ABUSE))) {
         logger.atWarning().log(
-            "Registrar '%s' (IANA ID %s) is missing ABUSE contact",
+            "Registrar '%s' (IANA ID %s) is missing ABUSE contact.",
             registrar.getRegistrarId(), registrar.getIanaIdentifier());
       }
       builder.entitiesBuilder().addAll(registrarContacts);

core/src/main/java/google/registry/rdap/RdapSearchActionBase.java

@@ -202,7 +202,7 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
               desiredRegistrar.get());
     }
     List<T> queryResult =
-        jpaTm().query(builder.build()).setMaxResults(querySizeLimit).getResultList();
+        jpaTm().criteriaQuery(builder.build()).setMaxResults(querySizeLimit).getResultList();
     if (checkForVisibility) {
       return filterResourcesByVisibility(queryResult, querySizeLimit);
     } else {

core/src/main/java/google/registry/rdap/UpdateRegistrarRdapBaseUrlsAction.java

@@ -92,7 +92,7 @@ public final class UpdateRegistrarRdapBaseUrlsAction implements Runnable {
   UpdateRegistrarRdapBaseUrlsAction() {}
 
   private String loginAndGetId(HttpRequestFactory requestFactory, String tld) throws IOException {
-    logger.atInfo().log("Logging in to MoSAPI");
+    logger.atInfo().log("Logging in to MoSAPI.");
     HttpRequest request =
         requestFactory.buildGetRequest(new GenericUrl(String.format(LOGIN_URL, tld)));
     request.getHeaders().setBasicAuthentication(String.format("%s_ry", tld), password);
@@ -176,15 +176,14 @@ public final class UpdateRegistrarRdapBaseUrlsAction implements Runnable {
       } catch (Throwable e) {
         // Login failures are bad but not unexpected for certain TLDs. We shouldn't store those
         // but rather should only store useful Throwables.
-        logger.atWarning().log("Error logging in to MoSAPI server: " + e.getMessage());
+        logger.atWarning().withCause(e).log("Error logging in to MoSAPI server.");
         continue;
       }
       try {
         return getRdapBaseUrlsPerIanaIdWithTld(tld, id, requestFactory);
       } catch (Throwable throwable) {
-        logger.atWarning().log(
-            String.format(
-                "Error retrieving RDAP urls with TLD %s: %s", tld, throwable.getMessage()));
+        logger.atWarning().withCause(throwable).log(
+            "Error retrieving RDAP URLs for TLD '%s'.", tld);
         finalThrowable = throwable;
       }
     }
@@ -215,7 +214,7 @@ public final class UpdateRegistrarRdapBaseUrlsAction implements Runnable {
                           // If this registrar already has these values, skip it
                           if (registrar.getRdapBaseUrls().equals(baseUrls)) {
                             logger.atInfo().log(
-                                "No change in RdapBaseUrls for registrar %s (ianaId %s)",
+                                "No change in RdapBaseUrls for registrar %s (ianaId %s).",
                                 registrar.getRegistrarId(), ianaId);
                             return;
                           }

core/src/main/java/google/registry/rde/BrdaCopyAction.java

@@ -92,7 +92,7 @@ public final class BrdaCopyAction implements Runnable {
 
     long xmlLength = readXmlLength(xmlLengthFilename);
 
-    logger.atInfo().log("Writing %s and %s", rydeFile, sigFile);
+    logger.atInfo().log("Writing files '%s' and '%s'.", rydeFile, sigFile);
     try (InputStream gcsInput = gcsUtils.openInputStream(xmlFilename);
         InputStream ghostrydeDecoder = Ghostryde.decoder(gcsInput, stagingDecryptionKey);
         OutputStream rydeOut = gcsUtils.openOutputStream(rydeFile);

core/src/main/java/google/registry/rde/EscrowTaskRunner.java

@@ -88,7 +88,7 @@ class EscrowTaskRunner {
       final Duration interval) {
     Callable<Void> lockRunner =
         () -> {
-          logger.atInfo().log("TLD: %s", registry.getTld());
+          logger.atInfo().log("Performing escrow for TLD '%s'.", registry.getTld());
           DateTime startOfToday = clock.nowUtc().withTimeAtStartOfDay();
           DateTime nextRequiredRun =
               transactIfJpaTm(
@@ -100,7 +100,7 @@ class EscrowTaskRunner {
           if (nextRequiredRun.isAfter(startOfToday)) {
             throw new NoContentException("Already completed");
           }
-          logger.atInfo().log("Current cursor is: %s", nextRequiredRun);
+          logger.atInfo().log("Current cursor is: %s.", nextRequiredRun);
           task.runWithLock(nextRequiredRun);
           DateTime nextRun = nextRequiredRun.plus(interval);
           logger.atInfo().log("Rolling cursor forward to %s.", nextRun);

core/src/main/java/google/registry/rde/RdeStagingAction.java

@@ -225,7 +225,7 @@ public final class RdeStagingAction implements Runnable {
     ImmutableSetMultimap<String, PendingDeposit> pendings =
         manual ? getManualPendingDeposits() : getStandardPendingDeposits();
     if (pendings.isEmpty()) {
-      String message = "Nothing needs to be deposited";
+      String message = "Nothing needs to be deposited.";
       logger.atInfo().log(message);
       response.setStatus(SC_NO_CONTENT);
       response.setPayload(message);

core/src/main/java/google/registry/rde/RdeStagingReducer.java

@@ -99,12 +99,12 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
         };
     String lockName = String.format("RdeStaging %s", key.mode());
     if (!lockHandler.executeWithLocks(lockRunner, key.tld(), lockTimeout, lockName)) {
-      logger.atWarning().log("Lock in use: %s", lockName);
+      logger.atWarning().log("Lock '%s' in use.", lockName);
     }
   }
 
   private void reduceWithLock(final PendingDeposit key, Iterator<DepositFragment> fragments) {
-    logger.atInfo().log("RdeStagingReducer %s", key);
+    logger.atInfo().log("RdeStagingReducer %s.", key);
 
     // Normally this is done by BackendServlet but it's not present in MapReduceServlet.
     Security.addProvider(new BouncyCastleProvider());
@@ -140,8 +140,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
     XjcRdeHeader header;
 
     // Write a gigantic XML file to GCS. We'll start by opening encrypted out/err file handles.
-
-    logger.atInfo().log("Writing %s and %s", xmlFilename, xmlLengthFilename);
+    logger.atInfo().log("Writing files '%s' and '%s'.", xmlFilename, xmlLengthFilename);
     try (OutputStream gcsOutput = gcsUtils.openOutputStream(xmlFilename);
         OutputStream lengthOutput = gcsUtils.openOutputStream(xmlLengthFilename);
         OutputStream ghostrydeEncoder = Ghostryde.encoder(gcsOutput, stagingKey, lengthOutput);
@@ -189,7 +188,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
     //
     // This will be sent to ICANN once we're done uploading the big XML to the escrow provider.
     if (mode == RdeMode.FULL) {
-      logger.atInfo().log("Writing %s", reportFilename);
+      logger.atInfo().log("Writing report file '%s'.", reportFilename);
       try (OutputStream gcsOutput = gcsUtils.openOutputStream(reportFilename);
           OutputStream ghostrydeEncoder = Ghostryde.encoder(gcsOutput, stagingKey)) {
         counter.makeReport(id, watermark, header, revision).marshal(ghostrydeEncoder, UTF_8);
@@ -200,7 +199,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
 
     // Now that we're done, kick off RdeUploadAction and roll the cursor forward.
     if (key.manual()) {
-      logger.atInfo().log("Manual operation; not advancing cursor or enqueuing upload task");
+      logger.atInfo().log("Manual operation; not advancing cursor or enqueuing upload task.");
       return;
     }
     tm().transact(
@@ -225,7 +224,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
                   key);
               tm().put(Cursor.create(key.cursor(), newPosition, registry));
               logger.atInfo().log(
-                  "Rolled forward %s on %s cursor to %s", key.cursor(), tld, newPosition);
+                  "Rolled forward %s on %s cursor to %s.", key.cursor(), tld, newPosition);
               RdeRevision.saveRevision(tld, watermark, mode, revision);
               if (mode == RdeMode.FULL) {
                 taskQueueUtils.enqueue(

core/src/main/java/google/registry/rde/RdeUploadAction.java

@@ -231,13 +231,13 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
                     .setFileMetadata(name, xmlLength, watermark)
                     .build()) {
             long bytesCopied = ByteStreams.copy(ghostrydeDecoder, rydeEncoder);
-            logger.atInfo().log("uploaded %,d bytes: %s", bytesCopied, rydeFilename);
+          logger.atInfo().log("Uploaded %,d bytes to path '%s'.", bytesCopied, rydeFilename);
           }
         String sigFilename = name + ".sig";
         byte[] signature = sigOut.toByteArray();
         gcsUtils.createFromBytes(BlobId.of(bucket, sigFilename), signature);
         ftpChan.get().put(new ByteArrayInputStream(signature), sigFilename);
-        logger.atInfo().log("uploaded %,d bytes: %s", signature.length, sigFilename);
+        logger.atInfo().log("Uploaded %,d bytes to path '%s'.", signature.length, sigFilename);
       }
     }
   }

core/src/main/java/google/registry/reporting/ReportingModule.java

@@ -14,7 +14,6 @@
 
 package google.registry.reporting;
 
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.RequestParameters.extractOptionalParameter;
 import static google.registry.request.RequestParameters.extractRequiredParameter;
 
@@ -56,9 +55,6 @@ public class ReportingModule {
   /** The request parameter specifying the jobId for a running Dataflow pipeline. */
   public static final String PARAM_JOB_ID = "jobId";
 
-  /** The request parameter for specifying which database reporting actions should read from. */
-  public static final String DATABASE = "database";
-
   /** Provides the Cloud Dataflow jobId for a pipeline. */
   @Provides
   @Parameter(PARAM_JOB_ID)
@@ -66,14 +62,6 @@ public class ReportingModule {
     return extractRequiredParameter(req, PARAM_JOB_ID);
   }
 
-  /** Provides the database for the pipeline to read from. */
-  @Provides
-  @Parameter(DATABASE)
-  static String provideDatabase(HttpServletRequest req) {
-    Optional<String> optionalDatabase = extractOptionalParameter(req, DATABASE);
-    return optionalDatabase.orElse(tm().isOfy() ? "DATASTORE" : "CLOUD_SQL");
-  }
-
   /** Extracts an optional YearMonth in yyyy-MM format from the request. */
   @Provides
   @Parameter(PARAM_YEAR_MONTH)

core/src/main/java/google/registry/reporting/billing/CopyDetailReportsAction.java

@@ -91,7 +91,7 @@ public final class CopyDetailReportsAction implements Runnable {
               .filter(objectName -> objectName.startsWith(BillingModule.DETAIL_REPORT_PREFIX))
               .collect(ImmutableList.toImmutableList());
     } catch (IOException e) {
-      logger.atSevere().withCause(e).log("Copying registrar detail report failed");
+      logger.atSevere().withCause(e).log("Copying registrar detail report failed.");
       response.setStatus(SC_INTERNAL_SERVER_ERROR);
       response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
       response.setPayload(String.format("Failure, encountered %s", e.getMessage()));
@@ -106,12 +106,12 @@ public final class CopyDetailReportsAction implements Runnable {
       Optional<Registrar> registrar = Registrar.loadByRegistrarId(registrarId);
       if (!registrar.isPresent()) {
         logger.atWarning().log(
-            "Registrar %s not found in database for file %s", registrar, detailReportName);
+            "Registrar %s not found in database for file '%s'.", registrar, detailReportName);
         continue;
       }
       String driveFolderId = registrar.get().getDriveFolderId();
       if (driveFolderId == null) {
-        logger.atWarning().log("Drive folder id not found for registrar %s", registrarId);
+        logger.atWarning().log("Drive folder id not found for registrar '%s'.", registrarId);
         continue;
       }
       // Attempt to copy each detail report to its associated registrar's drive folder.

core/src/main/java/google/registry/reporting/billing/GenerateInvoicesAction.java

@@ -15,10 +15,9 @@
 package google.registry.reporting.billing;
 
 import static google.registry.beam.BeamUtils.createJobName;
+import static google.registry.model.common.DatabaseMigrationStateSchedule.PrimaryDatabase.CLOUD_SQL;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
-import static google.registry.reporting.ReportingModule.DATABASE;
 import static google.registry.reporting.ReportingUtils.enqueueBeamReportingTask;
-import static google.registry.reporting.billing.BillingModule.PARAM_SHOULD_PUBLISH;
 import static google.registry.request.Action.Method.POST;
 import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
@@ -31,9 +30,11 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
 import google.registry.config.RegistryConfig.Config;
+import google.registry.model.common.DatabaseMigrationStateSchedule.PrimaryDatabase;
 import google.registry.reporting.ReportingModule;
 import google.registry.request.Action;
 import google.registry.request.Parameter;
+import google.registry.request.RequestParameters;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.util.Clock;
@@ -72,7 +73,7 @@ public class GenerateInvoicesAction implements Runnable {
   private final Clock clock;
   private final Response response;
   private final Dataflow dataflow;
-  private final String database;
+  private final PrimaryDatabase database;
 
   @Inject
   GenerateInvoicesAction(
@@ -81,8 +82,8 @@ public class GenerateInvoicesAction implements Runnable {
       @Config("beamStagingBucketUrl") String stagingBucketUrl,
       @Config("billingBucketUrl") String billingBucketUrl,
       @Config("invoiceFilePrefix") String invoiceFilePrefix,
-      @Parameter(PARAM_SHOULD_PUBLISH) boolean shouldPublish,
-      @Parameter(DATABASE) String database,
+      @Parameter(BillingModule.PARAM_SHOULD_PUBLISH) boolean shouldPublish,
+      @Parameter(RequestParameters.PARAM_DATABASE) PrimaryDatabase database,
       YearMonth yearMonth,
       BillingEmailUtils emailUtils,
       Clock clock,
@@ -93,7 +94,7 @@ public class GenerateInvoicesAction implements Runnable {
     this.stagingBucketUrl = stagingBucketUrl;
     // When generating the invoices using Cloud SQL before database cutover, save the reports in a
     // separate bucket so that it does not overwrite the Datastore invoices.
-    if (tm().isOfy() && database.equals("CLOUD_SQL")) {
+    if (tm().isOfy() && database.equals(CLOUD_SQL)) {
       billingBucketUrl = billingBucketUrl.concat("-sql");
     }
     this.billingBucketUrl = billingBucketUrl;
@@ -110,7 +111,7 @@ public class GenerateInvoicesAction implements Runnable {
   @Override
   public void run() {
     response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
-    logger.atInfo().log("Launching invoicing pipeline for %s", yearMonth);
+    logger.atInfo().log("Launching invoicing pipeline for %s.", yearMonth);
     try {
       LaunchFlexTemplateParameter parameter =
           new LaunchFlexTemplateParameter()
@@ -124,7 +125,7 @@ public class GenerateInvoicesAction implements Runnable {
                       "invoiceFilePrefix",
                       invoiceFilePrefix,
                       "database",
-                      database,
+                      database.name(),
                       "billingBucketUrl",
                       billingBucketUrl));
       LaunchFlexTemplateResponse launchResponse =
@@ -148,14 +149,13 @@ public class GenerateInvoicesAction implements Runnable {
                 yearMonth.toString());
         enqueueBeamReportingTask(PublishInvoicesAction.PATH, beamTaskParameters);
       }
+      response.setStatus(SC_OK);
+      response.setPayload(String.format("Launched invoicing pipeline: %s", jobId));
     } catch (IOException e) {
-      logger.atWarning().withCause(e).log("Template Launch failed");
-      emailUtils.sendAlertEmail(String.format("Template Launch failed due to %s", e.getMessage()));
+      logger.atWarning().withCause(e).log("Template Launch failed.");
+      emailUtils.sendAlertEmail(String.format("Pipeline Launch failed due to %s", e.getMessage()));
       response.setStatus(SC_INTERNAL_SERVER_ERROR);
-      response.setPayload(String.format("Template launch failed: %s", e.getMessage()));
-      return;
+      response.setPayload(String.format("Pipeline launch failed: %s", e.getMessage()));
     }
-    response.setStatus(SC_OK);
-    response.setPayload("Launched dataflow template.");
   }
 }

core/src/main/java/google/registry/reporting/icann/ActivityReportingQueryBuilder.java

@@ -14,8 +14,8 @@
 
 package google.registry.reporting.icann;
 
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.reporting.icann.IcannReportingModule.DATASTORE_EXPORT_DATA_SET;
-import static google.registry.reporting.icann.IcannReportingModule.ICANN_REPORTING_DATA_SET;
 import static google.registry.reporting.icann.QueryBuilderUtils.getQueryFromFile;
 import static google.registry.reporting.icann.QueryBuilderUtils.getTableName;
 
@@ -38,6 +38,7 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
   static final String EPP_METRICS = "epp_metrics";
   static final String WHOIS_COUNTS = "whois_counts";
   static final String ACTIVITY_REPORT_AGGREGATION = "activity_report_aggregation";
+  final String BIGQUERY_DATA_SET = tm().isOfy() ? "icann_reporting" : "cloud_sql_icann_reporting";
 
   @Inject
   @Config("projectId")
@@ -53,7 +54,7 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
   public String getReportQuery(YearMonth yearMonth) {
     return String.format(
         "#standardSQL\nSELECT * FROM `%s.%s.%s`",
-        projectId, ICANN_REPORTING_DATA_SET, getTableName(ACTIVITY_REPORT_AGGREGATION, yearMonth));
+        projectId, BIGQUERY_DATA_SET, getTableName(ACTIVITY_REPORT_AGGREGATION, yearMonth));
   }
 
   /** Sets the month we're doing activity reporting for, and returns the view query map. */
@@ -65,12 +66,20 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
     LocalDate lastDayOfMonth = yearMonth.toLocalDate(1).plusMonths(1).minusDays(1);
 
     ImmutableMap.Builder<String, String> queriesBuilder = ImmutableMap.builder();
-    String operationalRegistrarsQuery =
-        SqlTemplate.create(getQueryFromFile("registrar_operating_status.sql"))
-            .put("PROJECT_ID", projectId)
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .put("REGISTRAR_TABLE", "Registrar")
-            .build();
+    String operationalRegistrarsQuery;
+    if (tm().isOfy()) {
+      operationalRegistrarsQuery =
+          SqlTemplate.create(getQueryFromFile("registrar_operating_status.sql"))
+              .put("PROJECT_ID", projectId)
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
+              .put("REGISTRAR_TABLE", "Registrar")
+              .build();
+    } else {
+      operationalRegistrarsQuery =
+          SqlTemplate.create(getQueryFromFile("cloud_sql_registrar_operating_status.sql"))
+              .put("PROJECT_ID", projectId)
+              .build();
+    }
     queriesBuilder.put(
         getTableName(REGISTRAR_OPERATING_STATUS, yearMonth), operationalRegistrarsQuery);
 
@@ -93,7 +102,7 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
     String eppQuery =
         SqlTemplate.create(getQueryFromFile("epp_metrics.sql"))
             .put("PROJECT_ID", projectId)
-            .put("ICANN_REPORTING_DATA_SET", ICANN_REPORTING_DATA_SET)
+            .put("ICANN_REPORTING_DATA_SET", BIGQUERY_DATA_SET)
             .put("MONTHLY_LOGS_TABLE", getTableName(MONTHLY_LOGS, yearMonth))
             // All metadata logs for reporting come from google.registry.flows.FlowReporter.
             .put(
@@ -105,25 +114,35 @@ public final class ActivityReportingQueryBuilder implements QueryBuilder {
     String whoisQuery =
         SqlTemplate.create(getQueryFromFile("whois_counts.sql"))
             .put("PROJECT_ID", projectId)
-            .put("ICANN_REPORTING_DATA_SET", ICANN_REPORTING_DATA_SET)
+            .put("ICANN_REPORTING_DATA_SET", BIGQUERY_DATA_SET)
             .put("MONTHLY_LOGS_TABLE", getTableName(MONTHLY_LOGS, yearMonth))
             .build();
     queriesBuilder.put(getTableName(WHOIS_COUNTS, yearMonth), whoisQuery);
 
-    String aggregateQuery =
-        SqlTemplate.create(getQueryFromFile("activity_report_aggregation.sql"))
+    SqlTemplate aggregateQuery =
+        SqlTemplate.create(
+                getQueryFromFile(
+                    tm().isOfy()
+                        ? "activity_report_aggregation.sql"
+                        : "cloud_sql_activity_report_aggregation.sql"))
             .put("PROJECT_ID", projectId)
-            .put("ICANN_REPORTING_DATA_SET", ICANN_REPORTING_DATA_SET)
             .put(
                 "REGISTRAR_OPERATING_STATUS_TABLE",
                 getTableName(REGISTRAR_OPERATING_STATUS, yearMonth))
+            .put("ICANN_REPORTING_DATA_SET", BIGQUERY_DATA_SET)
             .put("DNS_COUNTS_TABLE", getTableName(DNS_COUNTS, yearMonth))
             .put("EPP_METRICS_TABLE", getTableName(EPP_METRICS, yearMonth))
-            .put("WHOIS_COUNTS_TABLE", getTableName(WHOIS_COUNTS, yearMonth))
-            .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET)
-            .put("REGISTRY_TABLE", "Registry")
-            .build();
-    queriesBuilder.put(getTableName(ACTIVITY_REPORT_AGGREGATION, yearMonth), aggregateQuery);
+            .put("WHOIS_COUNTS_TABLE", getTableName(WHOIS_COUNTS, yearMonth));
+
+    if (tm().isOfy()) {
+      aggregateQuery =
+          aggregateQuery
+              .put("REGISTRY_TABLE", "Registry")
+              .put("DATASTORE_EXPORT_DATA_SET", DATASTORE_EXPORT_DATA_SET);
+    }
+
+    queriesBuilder.put(
+        getTableName(ACTIVITY_REPORT_AGGREGATION, yearMonth), aggregateQuery.build());
 
     return queriesBuilder.build();
   }

core/src/main/java/google/registry/reporting/icann/IcannHttpReporter.java

@@ -86,7 +86,8 @@ public class IcannHttpReporter {
 
     HttpResponse response = null;
     logger.atInfo().log(
-        "Sending report to %s with content length %d", uploadUrl, request.getContent().getLength());
+        "Sending report to %s with content length %d.",
+        uploadUrl, request.getContent().getLength());
     boolean success = true;
     try {
       response = request.execute();

core/src/main/java/google/registry/reporting/icann/IcannReportingModule.java

@@ -14,6 +14,7 @@
 
 package google.registry.reporting.icann;
 
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.RequestParameters.extractOptionalParameter;
 import static google.registry.request.RequestParameters.extractRequiredParameter;
 import static google.registry.request.RequestParameters.extractSetOfEnumParameters;
@@ -41,7 +42,8 @@ public final class IcannReportingModule {
 
   static final String PARAM_SUBDIR = "subdir";
   static final String PARAM_REPORT_TYPES = "reportTypes";
-  static final String ICANN_REPORTING_DATA_SET = "icann_reporting";
+  static final String ICANN_REPORTING_DATA_SET =
+      tm().isOfy() ? "icann_reporting" : "cloud_sql_icann_reporting";
   static final String DATASTORE_EXPORT_DATA_SET = "latest_datastore_export";
   static final String MANIFEST_FILE_NAME = "MANIFEST.txt";
 

core/src/main/java/google/registry/reporting/icann/IcannReportingStager.java

@@ -72,7 +72,7 @@ public class IcannReportingStager {
   /**
    * Creates and stores reports of a given type on GCS.
    *
-   * <p>This is factored out to facilitate choosing which reports to upload,
+   * <p>This is factored out to facilitate choosing which reports to upload.
    */
   ImmutableList<String> stageReports(YearMonth yearMonth, String subdir, ReportType reportType)
       throws Exception {
@@ -90,11 +90,11 @@ public class IcannReportingStager {
       createIntermediaryTableView(entry.getKey(), entry.getValue(), reportType);
     }
 
-    // Get an in-memory table of the aggregate query's result
+    // Get an in-memory table of the aggregate query's result.
     ImmutableTable<Integer, TableFieldSchema, Object> reportTable =
         bigquery.queryToLocalTableSync(queryBuilder.getReportQuery(yearMonth));
 
-    // Get report headers from the table schema and convert into CSV format
+    // Get report headers from the table schema and convert into CSV format.
     String headerRow = constructRow(getHeaders(reportTable.columnKeySet()));
 
     return (reportType == ReportType.ACTIVITY)
@@ -104,8 +104,8 @@ public class IcannReportingStager {
 
   private void createIntermediaryTableView(String queryName, String query, ReportType reportType)
       throws ExecutionException, InterruptedException {
-    // Later views depend on the results of earlier ones, so query everything synchronously
-    logger.atInfo().log("Generating intermediary view %s", queryName);
+    // Later views depend on the results of earlier ones, so query everything synchronously.
+    logger.atInfo().log("Generating intermediary view %s.", queryName);
     bigquery
         .startQuery(
             query,
@@ -262,7 +262,7 @@ public class IcannReportingStager {
     final BlobId gcsFilename =
         BlobId.of(reportingBucket, String.format("%s/%s", subdir, reportFilename));
     gcsUtils.createFromBytes(gcsFilename, reportBytes);
-    logger.atInfo().log("Wrote %d bytes to file location %s", reportBytes.length, gcsFilename);
+    logger.atInfo().log("Wrote %d bytes to file location '%s'.", reportBytes.length, gcsFilename);
     return reportFilename;
   }
 
@@ -273,6 +273,6 @@ public class IcannReportingStager {
     StringBuilder manifestString = new StringBuilder();
     filenames.forEach((filename) -> manifestString.append(filename).append("\n"));
     gcsUtils.createFromBytes(gcsFilename, manifestString.toString().getBytes(UTF_8));
-    logger.atInfo().log("Wrote %d filenames to manifest at %s", filenames.size(), gcsFilename);
+    logger.atInfo().log("Wrote %d filenames to manifest at '%s'.", filenames.size(), gcsFilename);
   }
 }

core/src/main/java/google/registry/reporting/icann/IcannReportingStagingAction.java

@@ -118,7 +118,7 @@ public final class IcannReportingStagingAction implements Runnable {
             response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
             response.setPayload("Completed staging action.");
 
-            logger.atInfo().log("Enqueueing report upload :");
+            logger.atInfo().log("Enqueueing report upload.");
             TaskOptions uploadTask =
                 TaskOptions.Builder.withUrl(IcannReportingUploadAction.PATH)
                     .method(Method.POST)