Send expiring notification emails to admins if no tech emails are on file (#1387)

* Send emails to admin if tech emails are not present

* Improve test cases and comments

build.gradle

@@ -200,11 +200,28 @@ rootProject.ext {
   pyver = { exe ->
     try {
       ext.execInBash(
-          exe + " -c 'import sys; print(sys.hexversion)'", "/") as Integer
+          exe + " -c 'import sys; print(sys.hexversion)' 2>/dev/null",
+          "/") as Integer
     } catch (org.gradle.process.internal.ExecException e) {
       return -1;
     }
   }
+
+  // Return the path to a usable python3 executable.
+  getPythonExecutable = {
+    // 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) {
+      return 'python'
+    } else if (pyver('/usr/bin/python3') >= MIN_PY_VER) {
+      return '/usr/bin/python3'
+    } else {
+      throw new GradleException("No usable Python version found (build " +
+                                "requires at least python 3.7.3)");
+    }
+  }
 }
 
 task runPresubmits(type: Exec) {
@@ -212,18 +229,7 @@ task runPresubmits(type: Exec) {
   args('config/presubmits.py')
 
   doFirst {
-    // 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)");
-    }
+    executable getPythonExecutable()
   }
 }
 
@@ -438,9 +444,10 @@ rootProject.ext {
             ? "${rootDir}/.."
             : rootDir
     def formatDiffScript = "${scriptDir}/google-java-format-git-diff.sh"
+    def pythonExe = getPythonExecutable()
 
     return ext.execInBash(
-        "${formatDiffScript} ${action}", "${workingDir}")
+        "PYTHON=${pythonExe} ${formatDiffScript} ${action}", "${workingDir}")
   }
 }
 
@@ -448,18 +455,23 @@ rootProject.ext {
 // Note that this task checks modified Java files in the entire repository.
 task javaIncrementalFormatCheck {
   doLast {
-    def checkResult = invokeJavaDiffFormatScript("check")
-    if (checkResult == 'true') {
-      throw new IllegalStateException(
-          "Some Java files need to be reformatted. You may use the "
-              + "'javaIncrementalFormatDryRun' task to review\n "
-              + "the changes, or the 'javaIncrementalFormatApply' task "
-              + "to reformat.")
-    } else if (checkResult != 'false') {
-      throw new RuntimeException(
-          "Failed to invoke format check script:\n" + checkResult)
+    // We can only do this in a git tree.
+    if (new File("${rootDir}/.git").exists()) {
+      def checkResult = invokeJavaDiffFormatScript("check")
+      if (checkResult == 'true') {
+        throw new IllegalStateException(
+            "Some Java files need to be reformatted. You may use the "
+                + "'javaIncrementalFormatDryRun' task to review\n "
+                + "the changes, or the 'javaIncrementalFormatApply' task "
+                + "to reformat.")
+      } else if (checkResult != 'false') {
+        throw new RuntimeException(
+            "Failed to invoke format check script:\n" + checkResult)
+      }
+      println("Incremental Java format check ok.")
+    } else {
+      println("Omitting format check: not in a git directory.")
     }
-    println("Incremental Java format check ok.")
   }
 }
 

core/build.gradle

@@ -705,7 +705,11 @@ createToolTask(
 
 
 createToolTask(
-        'jpaDemoPipeline', 'google.registry.beam.common.JpaDemoPipeline')
+    'jpaDemoPipeline', 'google.registry.beam.common.JpaDemoPipeline')
+
+createToolTask(
+    'createSyntheticHistoryEntries',
+    'google.registry.tools.javascrap.CreateSyntheticHistoryEntriesPipeline')
 
 project.tasks.create('initSqlPipeline', JavaExec) {
   main = 'google.registry.beam.initsql.InitSqlPipeline'

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

@@ -113,9 +113,11 @@ public class SendExpiringCertificateNotificationEmailAction implements Runnable
   }
 
   /**
-   * Returns a list of registrars that should receive expiring notification emails. There are two
-   * certificates that should be considered (the main certificate and failOver certificate). The
-   * registrars should receive notifications if one of the certificate checks returns true.
+   * Returns a list of registrars that should receive expiring notification emails.
+   *
+   * <p>There are two certificates that should be considered (the main certificate and failOver
+   * certificate). The registrars should receive notifications if one of the certificate checks
+   * returns true.
    */
   @VisibleForTesting
   ImmutableList<RegistrarInfo> getRegistrarsWithExpiringCertificates() {
@@ -157,15 +159,17 @@ public class SendExpiringCertificateNotificationEmailAction implements Runnable
     }
     try {
       ImmutableSet<InternetAddress> recipients = getEmailAddresses(registrar, Type.TECH);
+      ImmutableSet<InternetAddress> ccs = getEmailAddresses(registrar, Type.ADMIN);
       Date expirationDate = certificateChecker.getCertificate(certificate.get()).getNotAfter();
       logger.atInfo().log(
-          "Registrar %s should receive an email that its %s SSL certificate will expire on %s.",
-          registrar.getRegistrarName(),
+          " %s SSL certificate of registrar '%s' will expire on %s.",
           certificateType.getDisplayName(),
+          registrar.getRegistrarName(),
           expirationDate.toString());
-      if (recipients.isEmpty()) {
+      if (recipients.isEmpty() && ccs.isEmpty()) {
         logger.atWarning().log(
-            "Registrar %s contains no email addresses to receive notification email.",
+            "Registrar %s contains no TECH nor ADMIN email addresses to receive notification"
+                + " email.",
             registrar.getRegistrarName());
         return false;
       }
@@ -180,7 +184,7 @@ public class SendExpiringCertificateNotificationEmailAction implements Runnable
                       expirationDate,
                       registrar.getRegistrarId()))
               .setRecipients(recipients)
-              .setCcs(getEmailAddresses(registrar, Type.ADMIN))
+              .setCcs(ccs)
               .build());
       /*
        * A duration time offset is used here to ensure that date comparison between two
@@ -249,30 +253,32 @@ public class SendExpiringCertificateNotificationEmailAction implements Runnable
   /** Sends notification emails to registrars with expiring certificates. */
   @VisibleForTesting
   int sendNotificationEmails() {
-    int emailsSent = 0;
+    int numEmailsSent = 0;
     for (RegistrarInfo registrarInfo : getRegistrarsWithExpiringCertificates()) {
       Registrar registrar = registrarInfo.registrar();
-      if (registrarInfo.isCertExpiring()) {
-        sendNotificationEmail(
-            registrar,
-            registrar.getLastExpiringCertNotificationSentDate(),
-            CertificateType.PRIMARY,
-            registrar.getClientCertificate());
-        emailsSent++;
+      if (registrarInfo.isCertExpiring()
+          && sendNotificationEmail(
+              registrar,
+              registrar.getLastExpiringCertNotificationSentDate(),
+              CertificateType.PRIMARY,
+              registrar.getClientCertificate())) {
+        numEmailsSent++;
       }
-      if (registrarInfo.isFailOverCertExpiring()) {
-        sendNotificationEmail(
-            registrar,
-            registrar.getLastExpiringFailoverCertNotificationSentDate(),
-            CertificateType.FAILOVER,
-            registrar.getFailoverClientCertificate());
-        emailsSent++;
+      if (registrarInfo.isFailOverCertExpiring()
+          && sendNotificationEmail(
+              registrar,
+              registrar.getLastExpiringFailoverCertNotificationSentDate(),
+              CertificateType.FAILOVER,
+              registrar.getFailoverClientCertificate())) {
+        numEmailsSent++;
       }
     }
-    return emailsSent;
+    return numEmailsSent;
   }
 
-  /** Returns a list of email addresses of the registrar that should receive a notification email */
+  /**
+   * Returns a list of email addresses of the registrar that should receive a notification email.
+   */
   @VisibleForTesting
   ImmutableSet<InternetAddress> getEmailAddresses(Registrar registrar, Type contactType) {
     ImmutableSortedSet<RegistrarContact> contacts = registrar.getContactsOfType(contactType);

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

@@ -21,6 +21,7 @@ import google.registry.config.CredentialModule;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.config.RegistryConfig.ConfigModule;
 import google.registry.persistence.PersistenceModule;
+import google.registry.persistence.PersistenceModule.BeamBulkQueryJpaTm;
 import google.registry.persistence.PersistenceModule.BeamJpaTm;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.transaction.JpaTransactionManager;
@@ -45,9 +46,19 @@ public interface RegistryPipelineComponent {
   @Config("projectId")
   String getProjectId();
 
+  /** Returns the regular {@link JpaTransactionManager} for general use. */
   @BeamJpaTm
   Lazy<JpaTransactionManager> getJpaTransactionManager();
 
+  /**
+   * Returns a {@link JpaTransactionManager} optimized for bulk loading multi-level JPA entities
+   * ({@link google.registry.model.domain.DomainBase} and {@link
+   * google.registry.model.domain.DomainHistory}). Please refer to {@link
+   * google.registry.model.bulkquery.BulkQueryEntities} for more information.
+   */
+  @BeamBulkQueryJpaTm
+  Lazy<JpaTransactionManager> getBulkQueryJpaTransactionManager();
+
   @Component.Builder
   interface Builder {
 

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

@@ -16,6 +16,7 @@ package google.registry.beam.common;
 
 import google.registry.beam.common.RegistryJpaIO.Write;
 import google.registry.config.RegistryEnvironment;
+import google.registry.persistence.PersistenceModule.JpaTransactionManagerType;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import java.util.Objects;
 import javax.annotation.Nullable;
@@ -44,6 +45,12 @@ public interface RegistryPipelineOptions extends GcpOptions {
 
   void setIsolationOverride(TransactionIsolationLevel isolationOverride);
 
+  @Description("The JPA Transaction Manager to use.")
+  @Default.Enum(value = "REGULAR")
+  JpaTransactionManagerType getJpaTransactionManagerType();
+
+  void setJpaTransactionManagerType(JpaTransactionManagerType jpaTransactionManagerType);
+
   @Description("The number of entities to write to the SQL database in one operation.")
   @Default.Integer(20)
   int getSqlWriteBatchSize();

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

@@ -49,10 +49,19 @@ public class RegistryPipelineWorkerInitializer implements JvmInitializer {
     }
     logger.atInfo().log("Setting up RegistryEnvironment %s.", environment);
     environment.setup();
-    Lazy<JpaTransactionManager> transactionManagerLazy =
-        toRegistryPipelineComponent(registryOptions).getJpaTransactionManager();
+    RegistryPipelineComponent registryPipelineComponent =
+        toRegistryPipelineComponent(registryOptions);
+    Lazy<JpaTransactionManager> transactionManagerLazy;
+    switch (registryOptions.getJpaTransactionManagerType()) {
+      case BULK_QUERY:
+        transactionManagerLazy = registryPipelineComponent.getBulkQueryJpaTransactionManager();
+        break;
+      case REGULAR:
+      default:
+        transactionManagerLazy = registryPipelineComponent.getJpaTransactionManager();
+    }
     TransactionManagerFactory.setJpaTmOnBeamWorker(transactionManagerLazy::get);
-    // Masquarade all threads as App Engine threads so we can create Ofy keys in the pipeline. Also
+    // Masquerade 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

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

@@ -14,6 +14,7 @@
 
 package google.registry.model;
 
+import google.registry.util.PreconditionsUtils;
 import javax.persistence.Access;
 import javax.persistence.AccessType;
 import javax.persistence.AttributeOverride;
@@ -49,4 +50,14 @@ public abstract class BackupGroupRoot extends ImmutableObject {
   public UpdateAutoTimestamp getUpdateTimestamp() {
     return updateTimestamp;
   }
+
+  /**
+   * Copies {@link #updateTimestamp} from another entity.
+   *
+   * <p>This method is for the few cases when {@code updateTimestamp} is copied between different
+   * types of entities. Use {@link #clone} for same-type copying.
+   */
+  protected void copyUpdateTimestamp(BackupGroupRoot other) {
+    this.updateTimestamp = PreconditionsUtils.checkArgumentNotNull(other, "other").updateTimestamp;
+  }
 }

core/src/main/java/google/registry/model/bulkquery/BulkQueryEntities.java

@@ -0,0 +1,109 @@
+// 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.model.bulkquery;
+
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.domain.DomainContent;
+import google.registry.model.domain.DomainHistory;
+import google.registry.model.domain.GracePeriod;
+import google.registry.model.domain.GracePeriod.GracePeriodHistory;
+import google.registry.model.domain.secdns.DelegationSignerData;
+import google.registry.model.domain.secdns.DomainDsDataHistory;
+import google.registry.model.host.HostResource;
+import google.registry.model.reporting.DomainTransactionRecord;
+import google.registry.persistence.VKey;
+import google.registry.persistence.transaction.JpaTransactionManager;
+
+/**
+ * Utilities for managing an alternative JPA entity model optimized for bulk loading multi-level
+ * entities such as {@link DomainBase} and {@link DomainHistory}.
+ *
+ * <p>In a bulk query for a multi-level JPA entity type, the JPA framework only generates a bulk
+ * query (SELECT * FROM table) for the base table. Then, for each row in the base table, additional
+ * queries are issued to load associated rows in child tables. This can be very slow when an entity
+ * type has multiple child tables.
+ *
+ * <p>We have defined an alternative entity model for {@code DomainBase} and {@code DomainHistory},
+ * where the base table as well as the child tables are mapped to single-level entity types. The
+ * idea is to load each of these types using a bulk query, and assemble them into the target type in
+ * memory in a pipeline. The main use case is Datastore-Cloud SQL validation during the Registry
+ * database migration, where we will need the full database snapshots frequently.
+ */
+public class BulkQueryEntities {
+  /**
+   * The JPA entity classes in persistence.xml to replace when creating the {@link
+   * JpaTransactionManager} for bulk query.
+   */
+  public static final ImmutableMap<String, String> JPA_ENTITIES_REPLACEMENTS =
+      ImmutableMap.of(
+          DomainBase.class.getCanonicalName(),
+          DomainBaseLite.class.getCanonicalName(),
+          DomainHistory.class.getCanonicalName(),
+          DomainHistoryLite.class.getCanonicalName());
+
+  /* The JPA entity classes that are not included in persistence.xml and need to be added to
+   * the {@link JpaTransactionManager} for bulk query.*/
+  public static final ImmutableList<String> JPA_ENTITIES_NEW =
+      ImmutableList.of(
+          DomainHost.class.getCanonicalName(), DomainHistoryHost.class.getCanonicalName());
+
+  public static DomainBase assembleDomainBase(
+      DomainBaseLite domainBaseLite,
+      ImmutableSet<GracePeriod> gracePeriods,
+      ImmutableSet<DelegationSignerData> delegationSignerData,
+      ImmutableSet<VKey<HostResource>> nsHosts) {
+    DomainBase.Builder builder = new DomainBase.Builder();
+    builder.copyFrom(domainBaseLite);
+    builder.setGracePeriods(gracePeriods);
+    builder.setDsData(delegationSignerData);
+    builder.setNameservers(nsHosts);
+    return builder.build();
+  }
+
+  public static DomainHistory assembleDomainHistory(
+      DomainHistoryLite domainHistoryLite,
+      ImmutableSet<DomainDsDataHistory> dsDataHistories,
+      ImmutableSet<VKey<HostResource>> domainHistoryHosts,
+      ImmutableSet<GracePeriodHistory> gracePeriodHistories,
+      ImmutableSet<DomainTransactionRecord> transactionRecords) {
+    DomainHistory.Builder builder = new DomainHistory.Builder();
+    builder.copyFrom(domainHistoryLite);
+    DomainContent rawDomainContent = domainHistoryLite.domainContent;
+    if (rawDomainContent != null) {
+      DomainContent newDomainContent =
+          domainHistoryLite
+              .domainContent
+              .asBuilder()
+              .setNameservers(domainHistoryHosts)
+              .setGracePeriods(
+                  gracePeriodHistories.stream()
+                      .map(GracePeriod::createFromHistory)
+                      .collect(toImmutableSet()))
+              .setDsData(
+                  dsDataHistories.stream()
+                      .map(DelegationSignerData::create)
+                      .collect(toImmutableSet()))
+              .build();
+      builder.setDomain(newDomainContent);
+    }
+    return builder.buildAndAssemble(
+        dsDataHistories, domainHistoryHosts, gracePeriodHistories, transactionRecords);
+  }
+}

core/src/main/java/google/registry/model/bulkquery/DomainBaseLite.java

@@ -0,0 +1,49 @@
+// 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.model.bulkquery;
+
+import google.registry.model.domain.DomainBase;
+import google.registry.model.domain.DomainContent;
+import google.registry.model.replay.SqlOnlyEntity;
+import google.registry.persistence.VKey;
+import google.registry.persistence.WithStringVKey;
+import javax.persistence.Access;
+import javax.persistence.AccessType;
+import javax.persistence.Entity;
+
+/**
+ * A 'light' version of {@link DomainBase} with only base table ("Domain") attributes, which allows
+ * fast bulk loading. They are used in in-memory assembly of {@code DomainBase} instances along with
+ * bulk-loaded child entities ({@code GracePeriod} etc). The in-memory assembly achieves much higher
+ * performance than loading {@code DomainBase} directly.
+ *
+ * <p>Please refer to {@link BulkQueryEntities} for more information.
+ */
+@Entity(name = "Domain")
+@WithStringVKey
+@Access(AccessType.FIELD)
+public class DomainBaseLite extends DomainContent implements SqlOnlyEntity {
+
+  @Override
+  @javax.persistence.Id
+  @Access(AccessType.PROPERTY)
+  public String getRepoId() {
+    return super.getRepoId();
+  }
+
+  public static VKey<DomainBaseLite> createVKey(String repoId) {
+    return VKey.createSql(DomainBaseLite.class, repoId);
+  }
+}

core/src/main/java/google/registry/model/bulkquery/DomainHistoryHost.java

@@ -0,0 +1,50 @@
+// 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.model.bulkquery;
+
+import google.registry.model.domain.DomainHistory.DomainHistoryId;
+import google.registry.model.host.HostResource;
+import google.registry.model.replay.SqlOnlyEntity;
+import google.registry.persistence.VKey;
+import java.io.Serializable;
+import javax.persistence.Access;
+import javax.persistence.AccessType;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.IdClass;
+
+/**
+ * A name server host referenced by a {@link google.registry.model.domain.DomainHistory} record.
+ * Please refer to {@link BulkQueryEntities} for usage.
+ */
+@Entity
+@Access(AccessType.FIELD)
+@IdClass(DomainHistoryHost.class)
+public class DomainHistoryHost implements Serializable, SqlOnlyEntity {
+
+  @Id private Long domainHistoryHistoryRevisionId;
+  @Id private String domainHistoryDomainRepoId;
+  @Id private String hostRepoId;
+
+  private DomainHistoryHost() {}
+
+  public DomainHistoryId getDomainHistoryId() {
+    return new DomainHistoryId(domainHistoryDomainRepoId, domainHistoryHistoryRevisionId);
+  }
+
+  public VKey<HostResource> getHostVKey() {
+    return VKey.create(HostResource.class, hostRepoId);
+  }
+}

core/src/main/java/google/registry/model/bulkquery/DomainHistoryLite.java

@@ -0,0 +1,122 @@
+// 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.model.bulkquery;
+
+import com.googlecode.objectify.Key;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.domain.DomainContent;
+import google.registry.model.domain.DomainHistory;
+import google.registry.model.domain.DomainHistory.DomainHistoryId;
+import google.registry.model.domain.Period;
+import google.registry.model.replay.SqlOnlyEntity;
+import google.registry.model.reporting.HistoryEntry;
+import google.registry.persistence.VKey;
+import javax.annotation.Nullable;
+import javax.persistence.Access;
+import javax.persistence.AccessType;
+import javax.persistence.AttributeOverride;
+import javax.persistence.AttributeOverrides;
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.IdClass;
+import javax.persistence.PostLoad;
+
+/**
+ * A 'light' version of {@link DomainHistory} with only base table ("DomainHistory") attributes,
+ * which allows fast bulk loading. They are used in in-memory assembly of {@code DomainHistory}
+ * instances along with bulk-loaded child entities ({@code GracePeriodHistory} etc). The in-memory
+ * assembly achieves much higher performance than loading {@code DomainHistory} directly.
+ *
+ * <p>Please refer to {@link BulkQueryEntities} for more information.
+ *
+ * <p>This class is adapted from {@link DomainHistory} by removing the {@code dsDataHistories},
+ * {@code gracePeriodHistories}, and {@code nsHosts} fields and associated methods.
+ */
+@Entity(name = "DomainHistory")
+@Access(AccessType.FIELD)
+@IdClass(DomainHistoryId.class)
+public class DomainHistoryLite extends HistoryEntry implements SqlOnlyEntity {
+
+  // Store DomainContent instead of DomainBase so we don't pick up its @Id
+  // Nullable for the sake of pre-Registry-3.0 history objects
+  @Nullable DomainContent domainContent;
+
+  @Id
+  @Access(AccessType.PROPERTY)
+  public String getDomainRepoId() {
+    // We need to handle null case here because Hibernate sometimes accesses this method before
+    // parent gets initialized
+    return parent == null ? null : parent.getName();
+  }
+
+  /** This method is private because it is only used by Hibernate. */
+  @SuppressWarnings("unused")
+  private void setDomainRepoId(String domainRepoId) {
+    parent = Key.create(DomainBase.class, domainRepoId);
+  }
+
+  @Override
+  @Nullable
+  @Access(AccessType.PROPERTY)
+  @AttributeOverrides({
+    @AttributeOverride(name = "unit", column = @Column(name = "historyPeriodUnit")),
+    @AttributeOverride(name = "value", column = @Column(name = "historyPeriodValue"))
+  })
+  public Period getPeriod() {
+    return super.getPeriod();
+  }
+
+  /**
+   * For transfers, the id of the other registrar.
+   *
+   * <p>For requests and cancels, the other registrar is the losing party (because the registrar
+   * sending the EPP transfer command is the gaining party). For approves and rejects, the other
+   * registrar is the gaining party.
+   */
+  @Nullable
+  @Access(AccessType.PROPERTY)
+  @Column(name = "historyOtherRegistrarId")
+  @Override
+  public String getOtherRegistrarId() {
+    return super.getOtherRegistrarId();
+  }
+
+  @Id
+  @Column(name = "historyRevisionId")
+  @Access(AccessType.PROPERTY)
+  @Override
+  public long getId() {
+    return super.getId();
+  }
+
+  /** The key to the {@link DomainBase} this is based off of. */
+  public VKey<DomainBase> getParentVKey() {
+    return VKey.create(DomainBase.class, getDomainRepoId());
+  }
+
+  @PostLoad
+  void postLoad() {
+    if (domainContent == null) {
+      return;
+    }
+    // See inline comments in DomainHistory.postLoad for reasons for the following lines.
+    if (domainContent.getDomainName() == null) {
+      domainContent = null;
+    } else if (domainContent.getRepoId() == null) {
+      domainContent.setRepoId(parent.getName());
+    }
+  }
+}

core/src/main/java/google/registry/model/bulkquery/DomainHost.java

@@ -0,0 +1,46 @@
+// 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.model.bulkquery;
+
+import google.registry.model.host.HostResource;
+import google.registry.model.replay.SqlOnlyEntity;
+import google.registry.persistence.VKey;
+import java.io.Serializable;
+import javax.persistence.Access;
+import javax.persistence.AccessType;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.IdClass;
+
+/** A name server host of a domain. Please refer to {@link BulkQueryEntities} for usage. */
+@Entity
+@Access(AccessType.FIELD)
+@IdClass(DomainHost.class)
+public class DomainHost implements Serializable, SqlOnlyEntity {
+
+  @Id private String domainRepoId;
+
+  @Id private String hostRepoId;
+
+  DomainHost() {}
+
+  public String getDomainRepoId() {
+    return domainRepoId;
+  }
+
+  public VKey<HostResource> getHostVKey() {
+    return VKey.create(HostResource.class, hostRepoId);
+  }
+}

core/src/main/java/google/registry/model/domain/DomainBase.java

@@ -189,6 +189,7 @@ public class DomainBase extends DomainContent
     }
 
     public Builder copyFrom(DomainContent domainContent) {
+      this.getInstance().copyUpdateTimestamp(domainContent);
       return this.setAuthInfo(domainContent.getAuthInfo())
           .setAutorenewPollMessage(domainContent.getAutorenewPollMessage())
           .setAutorenewBillingEvent(domainContent.getAutorenewBillingEvent())

core/src/main/java/google/registry/model/domain/DomainHistory.java

@@ -216,6 +216,10 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
     return super.getId();
   }
 
+  public DomainHistoryId getDomainHistoryId() {
+    return new DomainHistoryId(getDomainRepoId(), getId());
+  }
+
   /** Returns keys to the {@link HostResource} that are the nameservers for the domain. */
   public Set<VKey<HostResource>> getNsHosts() {
     return nsHosts;
@@ -314,6 +318,8 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
           nullToEmptyImmutableCopy(domainHistory.domainContent.getGracePeriods()).stream()
               .map(gracePeriod -> GracePeriodHistory.createFrom(domainHistory.id, gracePeriod))
               .collect(toImmutableSet());
+    } else {
+      domainHistory.nsHosts = ImmutableSet.of();
     }
   }
 
@@ -393,8 +399,16 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
       if (domainContent == null) {
         return this;
       }
+      // TODO(b/203609982): if actual type of domainContent is DomainBase, convert to DomainContent
+      // Note: a DomainHistory fetched by JPA has DomainContent in this field. Allowing DomainBase
+      // in the setter makes equality checks messy.
       getInstance().domainContent = domainContent;
-      return super.setParent(domainContent);
+      if (domainContent instanceof DomainBase) {
+        super.setParent(domainContent);
+      } else {
+        super.setParent(Key.create(DomainBase.class, domainContent.getRepoId()));
+      }
+      return this;
     }
 
     public Builder setDomainRepoId(String domainRepoId) {
@@ -412,5 +426,19 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
       fillAuxiliaryFieldsFromDomain(instance);
       return instance;
     }
+
+    public DomainHistory buildAndAssemble(
+        ImmutableSet<DomainDsDataHistory> dsDataHistories,
+        ImmutableSet<VKey<HostResource>> domainHistoryHosts,
+        ImmutableSet<GracePeriodHistory> gracePeriodHistories,
+        ImmutableSet<DomainTransactionRecord> transactionRecords) {
+      DomainHistory instance = super.build();
+      instance.dsDataHistories = dsDataHistories;
+      instance.nsHosts = domainHistoryHosts;
+      instance.gracePeriodHistories = gracePeriodHistories;
+      instance.domainTransactionRecords = transactionRecords;
+      instance.hashCode = null;
+      return instance;
+    }
   }
 }

core/src/main/java/google/registry/model/domain/GracePeriod.java

@@ -22,8 +22,10 @@ import com.google.common.annotations.VisibleForTesting;
 import com.googlecode.objectify.annotation.Embed;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.billing.BillingEvent.Recurring;
+import google.registry.model.domain.DomainHistory.DomainHistoryId;
 import google.registry.model.domain.rgp.GracePeriodStatus;
 import google.registry.model.replay.DatastoreAndSqlEntity;
+import google.registry.model.replay.SqlOnlyEntity;
 import google.registry.persistence.BillingVKey.BillingEventVKey;
 import google.registry.persistence.BillingVKey.BillingRecurrenceVKey;
 import google.registry.persistence.VKey;
@@ -202,7 +204,7 @@ public class GracePeriod extends GracePeriodBase implements DatastoreAndSqlEntit
   /** Entity class to represent a historic {@link GracePeriod}. */
   @Entity(name = "GracePeriodHistory")
   @Table(indexes = @Index(columnList = "domainRepoId"))
-  static class GracePeriodHistory extends GracePeriodBase {
+  public static class GracePeriodHistory extends GracePeriodBase implements SqlOnlyEntity {
     @Id Long gracePeriodHistoryRevisionId;
 
     /** ID for the associated {@link DomainHistory} entity. */
@@ -214,6 +216,10 @@ public class GracePeriod extends GracePeriodBase implements DatastoreAndSqlEntit
       return super.getGracePeriodId();
     }
 
+    public DomainHistoryId getDomainHistoryId() {
+      return new DomainHistoryId(getDomainRepoId(), domainHistoryRevisionId);
+    }
+
     static GracePeriodHistory createFrom(long historyRevisionId, GracePeriod gracePeriod) {
       GracePeriodHistory instance = new GracePeriodHistory();
       instance.gracePeriodHistoryRevisionId = allocateId();

core/src/main/java/google/registry/model/domain/secdns/DomainDsDataHistory.java

@@ -17,6 +17,7 @@ package google.registry.model.domain.secdns;
 import static google.registry.model.IdService.allocateId;
 
 import google.registry.model.domain.DomainHistory;
+import google.registry.model.domain.DomainHistory.DomainHistoryId;
 import google.registry.model.replay.SqlOnlyEntity;
 import javax.persistence.Access;
 import javax.persistence.AccessType;
@@ -53,6 +54,10 @@ public class DomainDsDataHistory extends DomainDsDataBase implements SqlOnlyEnti
     return instance;
   }
 
+  public DomainHistory.DomainHistoryId getDomainHistoryId() {
+    return new DomainHistoryId(getDomainRepoId(), domainHistoryRevisionId);
+  }
+
   @Override
   @Access(AccessType.PROPERTY)
   public String getDomainRepoId() {

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

@@ -22,6 +22,7 @@ import com.googlecode.objectify.annotation.Embed;
 import com.googlecode.objectify.annotation.Ignore;
 import google.registry.model.Buildable;
 import google.registry.model.ImmutableObject;
+import google.registry.model.domain.DomainHistory.DomainHistoryId;
 import google.registry.model.replay.DatastoreAndSqlEntity;
 import javax.persistence.Column;
 import javax.persistence.Entity;
@@ -49,7 +50,6 @@ public class DomainTransactionRecord extends ImmutableObject
 
   @Id
   @Ignore
-  @ImmutableObject.DoNotCompare
   @GeneratedValue(strategy = GenerationType.IDENTITY)
   @ImmutableObject.Insignificant
   Long id;
@@ -58,6 +58,14 @@ public class DomainTransactionRecord extends ImmutableObject
   @Column(nullable = false)
   String tld;
 
+  // The following two fields are exposed in this entity to support bulk-loading in Cloud SQL by the
+  // Datastore-SQL validation. They are excluded from equality check since they are not set in
+  // Datastore.
+  // TODO(b/203609782): post migration, decide whether to keep these two fields.
+  @Ignore @ImmutableObject.Insignificant String domainRepoId;
+
+  @Ignore @ImmutableObject.Insignificant Long historyRevisionId;
+
   /**
    * The time this Transaction takes effect (counting grace periods and other nuances).
    *
@@ -174,6 +182,10 @@ public class DomainTransactionRecord extends ImmutableObject
     }
   }
 
+  public DomainHistoryId getDomainHistoryId() {
+    return new DomainHistoryId(domainRepoId, historyRevisionId);
+  }
+
   public DateTime getReportingTime() {
     return reportingTime;
   }

core/src/main/java/google/registry/model/tmch/ClaimsEntry.java

@@ -15,7 +15,7 @@
 package google.registry.model.tmch;
 
 import google.registry.model.ImmutableObject;
-import google.registry.model.replay.NonReplicatedEntity;
+import google.registry.model.replay.SqlOnlyEntity;
 import java.io.Serializable;
 import javax.persistence.Column;
 import javax.persistence.Entity;
@@ -30,7 +30,7 @@ import javax.persistence.Id;
  * work.
  */
 @Entity(name = "ClaimsEntry")
-class ClaimsEntry extends ImmutableObject implements NonReplicatedEntity, Serializable {
+class ClaimsEntry extends ImmutableObject implements SqlOnlyEntity, Serializable {
   @Id private Long revisionId;
   @Id private String domainLabel;
 

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

@@ -0,0 +1,65 @@
+// 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.persistence;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Streams;
+import google.registry.model.bulkquery.BulkQueryEntities;
+import google.registry.persistence.transaction.JpaTransactionManager;
+import google.registry.persistence.transaction.JpaTransactionManagerImpl;
+import google.registry.util.Clock;
+import java.util.List;
+import javax.persistence.EntityManagerFactory;
+import org.hibernate.jpa.boot.internal.ParsedPersistenceXmlDescriptor;
+import org.hibernate.jpa.boot.spi.Bootstrap;
+
+/**
+ * Defines factory method for instantiating the bulk-query optimized {@link JpaTransactionManager}.
+ */
+public final class BulkQueryJpaFactory {
+
+  private BulkQueryJpaFactory() {}
+
+  static EntityManagerFactory createBulkQueryEntityManagerFactory(
+      ImmutableMap<String, String> cloudSqlConfigs) {
+    ParsedPersistenceXmlDescriptor descriptor =
+        PersistenceXmlUtility.getParsedPersistenceXmlDescriptor();
+
+    List<String> updatedManagedClasses =
+        Streams.concat(
+                descriptor.getManagedClassNames().stream(),
+                BulkQueryEntities.JPA_ENTITIES_NEW.stream())
+            .map(
+                name -> {
+                  if (BulkQueryEntities.JPA_ENTITIES_REPLACEMENTS.containsKey(name)) {
+                    return BulkQueryEntities.JPA_ENTITIES_REPLACEMENTS.get(name);
+                  }
+                  return name;
+                })
+            .collect(ImmutableList.toImmutableList());
+
+    descriptor.getManagedClassNames().clear();
+    descriptor.getManagedClassNames().addAll(updatedManagedClasses);
+
+    return Bootstrap.getEntityManagerFactoryBuilder(descriptor, cloudSqlConfigs).build();
+  }
+
+  public static JpaTransactionManager createBulkQueryJpaTransactionManager(
+      ImmutableMap<String, String> cloudSqlConfigs, Clock clock) {
+    return new JpaTransactionManagerImpl(
+        createBulkQueryEntityManagerFactory(cloudSqlConfigs), clock);
+  }
+}

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

@@ -152,13 +152,36 @@ public abstract class PersistenceModule {
   @Singleton
   @BeamPipelineCloudSqlConfigs
   static ImmutableMap<String, String> provideBeamPipelineCloudSqlConfigs(
-      @Config("beamCloudSqlJdbcUrl") String jdbcUrl,
-      @Config("beamCloudSqlInstanceConnectionName") String instanceConnectionName,
-      @DefaultHibernateConfigs ImmutableMap<String, String> defaultConfigs,
+      SqlCredentialStore credentialStore,
+      @Config("instanceConnectionNameOverride")
+          Optional<Provider<String>> instanceConnectionNameOverride,
       @Config("beamIsolationOverride")
-          Optional<Provider<TransactionIsolationLevel>> isolationOverride) {
-    return createPartialSqlConfigs(
-        jdbcUrl, instanceConnectionName, defaultConfigs, isolationOverride);
+          Optional<Provider<TransactionIsolationLevel>> isolationOverride,
+      @PartialCloudSqlConfigs ImmutableMap<String, String> cloudSqlConfigs) {
+    HashMap<String, String> overrides = Maps.newHashMap(cloudSqlConfigs);
+    // TODO(b/175700623): make sql username configurable from config file.
+    SqlCredential credential = credentialStore.getCredential(new RobotUser(RobotId.NOMULUS));
+    overrides.put(Environment.USER, credential.login());
+    overrides.put(Environment.PASS, credential.password());
+    // Override the default minimum which is tuned for the Registry server. A worker VM should
+    // release all connections if it no longer interacts with the database.
+    overrides.put(HIKARI_MINIMUM_IDLE, "0");
+    /**
+     * Disable Hikari's maxPoolSize limit check by setting it to an absurdly large number. The
+     * effective (and desirable) limit is the number of pipeline threads on the pipeline worker,
+     * which can be configured using pipeline options. See {@link RegistryPipelineOptions} for more
+     * information.
+     */
+    overrides.put(HIKARI_MAXIMUM_POOL_SIZE, String.valueOf(Integer.MAX_VALUE));
+    instanceConnectionNameOverride
+        .map(Provider::get)
+        .ifPresent(
+            instanceConnectionName ->
+                overrides.put(HIKARI_DS_CLOUD_SQL_INSTANCE, instanceConnectionName));
+    isolationOverride
+        .map(Provider::get)
+        .ifPresent(isolation -> overrides.put(Environment.ISOLATION, isolation.name()));
+    return ImmutableMap.copyOf(overrides);
   }
 
   @VisibleForTesting
@@ -230,37 +253,17 @@ public abstract class PersistenceModule {
   @Singleton
   @BeamJpaTm
   static JpaTransactionManager provideBeamJpaTm(
-      SqlCredentialStore credentialStore,
-      @Config("instanceConnectionNameOverride")
-          Optional<Provider<String>> instanceConnectionNameOverride,
-      @Config("beamIsolationOverride")
-          Optional<Provider<TransactionIsolationLevel>> isolationOverride,
-      @PartialCloudSqlConfigs ImmutableMap<String, String> cloudSqlConfigs,
-      Clock clock) {
-    HashMap<String, String> overrides = Maps.newHashMap(cloudSqlConfigs);
-    // TODO(b/175700623): make sql username configurable from config file.
-    SqlCredential credential = credentialStore.getCredential(new RobotUser(RobotId.NOMULUS));
-    overrides.put(Environment.USER, credential.login());
-    overrides.put(Environment.PASS, credential.password());
-    // Override the default minimum which is tuned for the Registry server. A worker VM should
-    // release all connections if it no longer interacts with the database.
-    overrides.put(HIKARI_MINIMUM_IDLE, "0");
-    /**
-     * Disable Hikari's maxPoolSize limit check by setting it to an absurdly large number. The
-     * effective (and desirable) limit is the number of pipeline threads on the pipeline worker,
-     * which can be configured using pipeline options. See {@link RegistryPipelineOptions} for more
-     * information.
-     */
-    overrides.put(HIKARI_MAXIMUM_POOL_SIZE, String.valueOf(Integer.MAX_VALUE));
-    instanceConnectionNameOverride
-        .map(Provider::get)
-        .ifPresent(
-            instanceConnectionName ->
-                overrides.put(HIKARI_DS_CLOUD_SQL_INSTANCE, instanceConnectionName));
-    isolationOverride
-        .map(Provider::get)
-        .ifPresent(isolation -> overrides.put(Environment.ISOLATION, isolation.name()));
-    return new JpaTransactionManagerImpl(create(overrides), clock);
+      @BeamPipelineCloudSqlConfigs ImmutableMap<String, String> beamCloudSqlConfigs, Clock clock) {
+    return new JpaTransactionManagerImpl(create(beamCloudSqlConfigs), clock);
+  }
+
+  @Provides
+  @Singleton
+  @BeamBulkQueryJpaTm
+  static JpaTransactionManager provideBeamBulkQueryJpaTm(
+      @BeamPipelineCloudSqlConfigs ImmutableMap<String, String> beamCloudSqlConfigs, Clock clock) {
+    return new JpaTransactionManagerImpl(
+        BulkQueryJpaFactory.createBulkQueryEntityManagerFactory(beamCloudSqlConfigs), clock);
   }
 
   @Provides
@@ -346,6 +349,17 @@ public abstract class PersistenceModule {
     }
   }
 
+  /** Types of {@link JpaTransactionManager JpaTransactionManagers}. */
+  public enum JpaTransactionManagerType {
+    /** The regular {@link JpaTransactionManager} for general use. */
+    REGULAR,
+    /**
+     * The {@link JpaTransactionManager} optimized for bulk loading multi-level JPA entities. Please
+     * see {@link google.registry.model.bulkquery.BulkQueryEntities} for more information.
+     */
+    BULK_QUERY
+  }
+
   /** Dagger qualifier for JDBC {@link Connection} with schema management privilege. */
   @Qualifier
   @Documented
@@ -357,11 +371,18 @@ public abstract class PersistenceModule {
   @interface AppEngineJpaTm {}
 
   /** Dagger qualifier for {@link JpaTransactionManager} used inside BEAM pipelines. */
-  // Note: @SocketFactoryJpaTm will be phased out in favor of this qualifier.
   @Qualifier
   @Documented
   public @interface BeamJpaTm {}
 
+  /**
+   * Dagger qualifier for {@link JpaTransactionManager} that uses an alternative entity model for
+   * faster bulk queries.
+   */
+  @Qualifier
+  @Documented
+  public @interface BeamBulkQueryJpaTm {}
+
   /** Dagger qualifier for {@link JpaTransactionManager} used for Nomulus tool. */
   @Qualifier
   @Documented

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

@@ -14,7 +14,6 @@
 
 package google.registry.reporting.icann;
 
-import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.net.MediaType.PLAIN_TEXT_UTF_8;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.POST;
@@ -46,7 +45,6 @@ import google.registry.util.SendEmailService;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Map;
-import java.util.concurrent.Callable;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import javax.inject.Inject;
@@ -78,6 +76,7 @@ public final class IcannReportingUploadAction implements Runnable {
   static final String PATH = "/_dr/task/icannReportingUpload";
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final String LOCK_NAME = "IcannReportingUploadAction";
 
   @Inject
   @Config("reportingBucket")
@@ -98,48 +97,33 @@ public final class IcannReportingUploadAction implements Runnable {
 
   @Override
   public void run() {
-    Runnable transactional =
-        () -> {
-          ImmutableMap.Builder<String, Boolean> reportSummaryBuilder = new ImmutableMap.Builder<>();
-
-          ImmutableMap<Cursor, String> cursors = loadCursors();
-
-          // If cursor time is before now, upload the corresponding report
-          cursors.entrySet().stream()
-              .filter(entry -> entry.getKey().getCursorTime().isBefore(clock.nowUtc()))
-              .forEach(
-                  entry -> {
-                    DateTime cursorTime = entry.getKey().getCursorTime();
-                    uploadReport(
-                        cursorTime,
-                        entry.getKey().getType(),
-                        entry.getValue(),
-                        reportSummaryBuilder);
-                  });
-          // Send email of which reports were uploaded
-          emailUploadResults(reportSummaryBuilder.build());
-          response.setStatus(SC_OK);
-          response.setContentType(PLAIN_TEXT_UTF_8);
-        };
-
-    Callable<Void> lockRunner =
-        () -> {
-          tm().transact(transactional);
-          return null;
-        };
-
-    String lockname = "IcannReportingUploadAction";
-    if (!lockHandler.executeWithLocks(lockRunner, null, Duration.standardHours(2), lockname)) {
-      throw new ServiceUnavailableException("Lock for IcannReportingUploadAction already in use");
+    if (!lockHandler.executeWithLocks(
+        this::runWithLock, null, Duration.standardHours(2), LOCK_NAME)) {
+      throw new ServiceUnavailableException(String.format("Lock for %s already in use", LOCK_NAME));
     }
   }
 
+  private Void runWithLock() {
+    ImmutableMap.Builder<String, Boolean> reportSummaryBuilder = new ImmutableMap.Builder<>();
+
+    ImmutableMap<Cursor, String> cursors = tm().transact(this::loadCursors);
+
+    // If cursor time is before now, upload the corresponding report
+    cursors.entrySet().stream()
+        .filter(entry -> entry.getKey().getCursorTime().isBefore(clock.nowUtc()))
+        .forEach(entry -> uploadReport(entry.getKey(), entry.getValue(), reportSummaryBuilder));
+    // Send email of which reports were uploaded
+    emailUploadResults(reportSummaryBuilder.build());
+    response.setStatus(SC_OK);
+    response.setContentType(PLAIN_TEXT_UTF_8);
+    return null;
+  }
+
   /** Uploads the report and rolls forward the cursor for that report. */
   private void uploadReport(
-      DateTime cursorTime,
-      CursorType cursorType,
-      String tldStr,
-      ImmutableMap.Builder<String, Boolean> reportSummaryBuilder) {
+      Cursor cursor, String tldStr, ImmutableMap.Builder<String, Boolean> reportSummaryBuilder) {
+    DateTime cursorTime = cursor.getCursorTime();
+    CursorType cursorType = cursor.getType();
     DateTime cursorTimeMinusMonth = cursorTime.withDayOfMonth(1).minusMonths(1);
     String reportSubdir =
         String.format(
@@ -150,17 +134,16 @@ public final class IcannReportingUploadAction implements Runnable {
         BlobId.of(reportingBucket, String.format("%s/%s", reportSubdir, filename));
     logger.atInfo().log("Reading ICANN report %s from bucket '%s'.", filename, reportingBucket);
     // Check that the report exists
-    try {
-      verifyFileExists(gcsFilename);
-    } catch (IllegalArgumentException e) {
+    if (!gcsUtils.existsAndNotEmpty(gcsFilename)) {
       String logMessage =
           String.format(
-              "Could not upload %s report for %s because file %s did not exist.",
-              cursorType, tldStr, filename);
+              "Could not upload %s report for %s because file %s (object %s in bucket %s) did not"
+                  + " exist.",
+              cursorType, tldStr, filename, gcsFilename.getName(), gcsFilename.getBucket());
       if (clock.nowUtc().dayOfMonth().get() == 1) {
-        logger.atInfo().withCause(e).log(logMessage + " This report may not have been staged yet.");
+        logger.atInfo().log(logMessage + " This report may not have been staged yet.");
       } else {
-        logger.atSevere().withCause(e).log(logMessage);
+        logger.atSevere().log(logMessage);
       }
       reportSummaryBuilder.put(filename, false);
       return;
@@ -179,7 +162,6 @@ public final class IcannReportingUploadAction implements Runnable {
     } catch (RuntimeException e) {
       logger.atWarning().withCause(e).log("Upload to %s failed.", gcsFilename);
     }
-    reportSummaryBuilder.put(filename, success);
 
     // Set cursor to first day of next month if the upload succeeded
     if (success) {
@@ -188,8 +170,24 @@ public final class IcannReportingUploadAction implements Runnable {
               cursorType,
               cursorTime.withTimeAtStartOfDay().withDayOfMonth(1).plusMonths(1),
               Registry.get(tldStr));
-      tm().put(newCursor);
+      // In order to keep the transactions short-lived, we load all of the cursors in a single
+      // transaction then later use per-cursor transactions when checking + saving the cursors. We
+      // run behind a lock so the cursors shouldn't be changed, but double check to be sure.
+      success =
+          tm().transact(
+                  () -> {
+                    Cursor fromDb = tm().transact(() -> tm().loadByEntity(cursor));
+                    if (!cursor.equals(fromDb)) {
+                      logger.atSevere().log(
+                          "Expected previously-loaded cursor %s to equal current cursor %s",
+                          cursor, fromDb);
+                      return false;
+                    }
+                    tm().put(newCursor);
+                    return true;
+                  });
     }
+    reportSummaryBuilder.put(filename, success);
   }
 
   private String getFileName(CursorType cursorType, DateTime cursorTime, String tld) {
@@ -303,13 +301,4 @@ public final class IcannReportingUploadAction implements Runnable {
       return ByteStreams.toByteArray(gcsInput);
     }
   }
-
-  private void verifyFileExists(BlobId gcsFilename) {
-    checkArgument(
-        gcsUtils.existsAndNotEmpty(gcsFilename),
-        "Object %s in bucket %s not found",
-        gcsFilename.getName(),
-        gcsFilename.getBucket());
-  }
-
 }

core/src/main/java/google/registry/tools/RegistryTool.java

@@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableMap;
 import google.registry.tools.javascrap.BackfillRegistryLocksCommand;
 import google.registry.tools.javascrap.BackfillSpec11ThreatMatchesCommand;
 import google.registry.tools.javascrap.DeleteContactByRoidCommand;
+import google.registry.tools.javascrap.HardDeleteHostCommand;
 import google.registry.tools.javascrap.PopulateNullRegistrarFieldsCommand;
 import google.registry.tools.javascrap.RemoveIpAddressCommand;
 import google.registry.tools.javascrap.ResaveAllTldsCommand;
@@ -86,6 +87,7 @@ public final class RegistryTool {
           .put("get_sql_credential", GetSqlCredentialCommand.class)
           .put("get_tld", GetTldCommand.class)
           .put("ghostryde", GhostrydeCommand.class)
+          .put("hard_delete_host", HardDeleteHostCommand.class)
           .put("hash_certificate", HashCertificateCommand.class)
           .put("import_datastore", ImportDatastoreCommand.class)
           .put("list_cursors", ListCursorsCommand.class)

core/src/main/java/google/registry/tools/RegistryToolComponent.java

@@ -43,6 +43,7 @@ import google.registry.request.Modules.UserServiceModule;
 import google.registry.tools.AuthModule.LocalCredentialModule;
 import google.registry.tools.javascrap.BackfillRegistryLocksCommand;
 import google.registry.tools.javascrap.DeleteContactByRoidCommand;
+import google.registry.tools.javascrap.HardDeleteHostCommand;
 import google.registry.util.UtilsModule;
 import google.registry.whois.NonCachingWhoisModule;
 import javax.annotation.Nullable;
@@ -124,6 +125,8 @@ interface RegistryToolComponent {
 
   void inject(GhostrydeCommand command);
 
+  void inject(HardDeleteHostCommand command);
+
   void inject(ImportDatastoreCommand command);
 
   void inject(ListCursorsCommand command);

core/src/main/java/google/registry/tools/javascrap/CreateSyntheticHistoryEntriesAction.java

@@ -108,6 +108,16 @@ public class CreateSyntheticHistoryEntriesAction implements Runnable {
     return jpaTm()
         .transact(
             () -> {
+              // Use READ COMMITTED isolation level so that any long-living queries don't cause
+              // collection of predicate locks to spiral out of control (as would happen with a
+              // SERIALIZABLE isolation level)
+              //
+              // NB: setting the isolation level inside the transaction only works for Postgres and
+              // will be reverted to the default once the transaction is committed.
+              jpaTm()
+                  .getEntityManager()
+                  .createNativeQuery("SET TRANSACTION ISOLATION LEVEL READ COMMITTED")
+                  .executeUpdate();
               // The class we're searching from is based on which parent type (e.g. Domain) we have
               Class<? extends HistoryEntry> historyClass =
                   getHistoryClassFromParent(resource.getClass());

core/src/main/java/google/registry/tools/javascrap/CreateSyntheticHistoryEntriesPipeline.java

@@ -0,0 +1,135 @@
+// 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.tools.javascrap;
+
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+
+import com.google.common.collect.ImmutableList;
+import dagger.Component;
+import google.registry.beam.common.RegistryJpaIO;
+import google.registry.beam.common.RegistryPipelineOptions;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.config.RegistryConfig.ConfigModule;
+import google.registry.model.EppResource;
+import google.registry.model.UpdateAutoTimestamp;
+import google.registry.model.UpdateAutoTimestamp.DisableAutoUpdateResource;
+import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.host.HostResource;
+import google.registry.model.reporting.HistoryEntry;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
+import google.registry.persistence.VKey;
+import java.io.Serializable;
+import javax.inject.Singleton;
+import javax.persistence.Entity;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.values.TypeDescriptor;
+
+/**
+ * Pipeline that creates a synthetic history entry for every {@link EppResource} in SQL at the
+ * current time.
+ *
+ * <p>The history entries in Datastore does not have the EPP resource embedded in them. Therefore
+ * after {@link google.registry.beam.initsql.InitSqlPipeline} runs, these fields will all be empty.
+ * This pipeline loads all EPP resources and for each of them creates a synthetic history entry that
+ * contains the resource and saves them back to SQL, so that they can be used in the RDE pipeline.
+ *
+ * <p>Note that this pipeline should only be run in a test environment right after the init SQL
+ * pipeline finishes, and no EPP update is being made to the system, otherwise there is no garuantee
+ * that the latest history entry for a given EPP resource does not already have the resource
+ * embedded within it.
+ *
+ * <p>To run the pipeline:
+ *
+ * <p><code>
+ * $ ./nom_build :core:cSHE --args="--region=us-central1
+ *   --runner=DataflowRunner
+ *   --registryEnvironment=CRASH
+ *   --project={project-id}
+ *   --workerMachineType=n2-standard-4"
+ * </code>
+ *
+ * @see google.registry.tools.javascrap.CreateSyntheticHistoryEntriesAction
+ */
+public class CreateSyntheticHistoryEntriesPipeline implements Serializable {
+
+  private static final ImmutableList<Class<? extends EppResource>> EPP_RESOURCE_CLASSES =
+      ImmutableList.of(DomainBase.class, ContactResource.class, HostResource.class);
+
+  private static final String HISTORY_REASON =
+      "Backfill EppResource history objects after initial backup to SQL";
+
+  static void setup(Pipeline pipeline, String registryAdminRegistrarId) {
+    for (Class<? extends EppResource> clazz : EPP_RESOURCE_CLASSES) {
+      pipeline
+          .apply(
+              String.format("Read all %s", clazz.getSimpleName()),
+              RegistryJpaIO.read(
+                  "SELECT id FROM %entity%"
+                      .replace("%entity%", clazz.getAnnotation(Entity.class).name()),
+                  String.class,
+                  repoId -> VKey.createSql(clazz, repoId)))
+          .apply(
+              String.format("Save a synthetic HistoryEntry for each %s", clazz),
+              MapElements.into(TypeDescriptor.of(Void.class))
+                  .via(
+                      (VKey<? extends EppResource> key) -> {
+                        jpaTm()
+                            .transact(
+                                () -> {
+                                  EppResource eppResource = jpaTm().loadByKey(key);
+                                  try (DisableAutoUpdateResource disable =
+                                      UpdateAutoTimestamp.disableAutoUpdate()) {
+                                    jpaTm()
+                                        .put(
+                                            HistoryEntry.createBuilderForResource(eppResource)
+                                                .setRegistrarId(registryAdminRegistrarId)
+                                                .setBySuperuser(true)
+                                                .setRequestedByRegistrar(false)
+                                                .setModificationTime(jpaTm().getTransactionTime())
+                                                .setReason(HISTORY_REASON)
+                                                .setType(HistoryEntry.Type.SYNTHETIC)
+                                                .build());
+                                  }
+                                });
+                        return null;
+                      }));
+    }
+  }
+
+  public static void main(String[] args) {
+    RegistryPipelineOptions options =
+        PipelineOptionsFactory.fromArgs(args).withValidation().as(RegistryPipelineOptions.class);
+    RegistryPipelineOptions.validateRegistryPipelineOptions(options);
+    options.setIsolationOverride(TransactionIsolationLevel.TRANSACTION_READ_COMMITTED);
+    String registryAdminRegistrarId =
+        DaggerCreateSyntheticHistoryEntriesPipeline_ConfigComponent.create()
+            .getRegistryAdminRegistrarId();
+
+    Pipeline pipeline = Pipeline.create(options);
+    setup(pipeline, registryAdminRegistrarId);
+    pipeline.run();
+  }
+
+  @Singleton
+  @Component(modules = ConfigModule.class)
+  interface ConfigComponent {
+
+    @Config("registryAdminClientId")
+    String getRegistryAdminRegistrarId();
+  }
+}

core/src/main/java/google/registry/tools/javascrap/HardDeleteHostCommand.java

@@ -0,0 +1,99 @@
+// 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.tools.javascrap;
+
+import static com.google.common.base.Verify.verify;
+import static google.registry.model.ofy.ObjectifyService.auditedOfy;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.Parameters;
+import com.google.common.collect.ImmutableList;
+import com.googlecode.objectify.Key;
+import google.registry.model.host.HostResource;
+import google.registry.model.index.EppResourceIndex;
+import google.registry.model.index.ForeignKeyIndex;
+import google.registry.tools.CommandWithRemoteApi;
+import google.registry.tools.ConfirmingCommand;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Deletes a {@link HostResource} by its ROID.
+ *
+ * <p>This deletes the host itself, everything in the same entity group including all {@link
+ * google.registry.model.reporting.HistoryEntry}s and {@link
+ * google.registry.model.poll.PollMessage}s, the {@link EppResourceIndex}, and the {@link
+ * ForeignKeyIndex} (if it exists).
+ *
+ * <p>DO NOT use this to hard-delete a host that is still in use on a domain. Bad things will
+ * happen.
+ */
+@Parameters(separators = " =", commandDescription = "Delete a host by its ROID.")
+public class HardDeleteHostCommand extends ConfirmingCommand implements CommandWithRemoteApi {
+
+  @Parameter(names = "--roid", description = "The ROID of the host to be deleted.")
+  String roid;
+
+  @Parameter(names = "--hostname", description = "The hostname, for verification.")
+  String hostname;
+
+  private ImmutableList<Key<Object>> toDelete;
+
+  @Override
+  protected void init() {
+    ofyTm()
+        .transact(
+            () -> {
+              Key<HostResource> targetKey = Key.create(HostResource.class, roid);
+              HostResource host = auditedOfy().load().key(targetKey).now();
+              verify(Objects.equals(host.getHostName(), hostname), "Hostname does not match");
+
+              List<Key<Object>> objectsInEntityGroup =
+                  auditedOfy().load().ancestor(host).keys().list();
+
+              Optional<ForeignKeyIndex<HostResource>> fki =
+                  Optional.ofNullable(
+                      auditedOfy().load().key(ForeignKeyIndex.createKey(host)).now());
+              if (!fki.isPresent()) {
+                System.out.println(
+                    "No ForeignKeyIndex exists, likely because resource is soft-deleted."
+                        + " Continuing.");
+              }
+
+              EppResourceIndex eppResourceIndex =
+                  auditedOfy().load().entity(EppResourceIndex.create(targetKey)).now();
+              verify(eppResourceIndex.getKey().equals(targetKey), "Wrong EppResource Index loaded");
+
+              ImmutableList.Builder<Key<Object>> toDeleteBuilder =
+                  new ImmutableList.Builder<Key<Object>>()
+                      .addAll(objectsInEntityGroup)
+                      .add(Key.create(eppResourceIndex));
+              fki.ifPresent(f -> toDeleteBuilder.add(Key.create(f)));
+              toDelete = toDeleteBuilder.build();
+
+              System.out.printf("\n\nAbout to delete %d entities with keys:\n", toDelete.size());
+              toDelete.forEach(System.out::println);
+            });
+  }
+
+  @Override
+  protected String execute() {
+    tm().transact(() -> auditedOfy().delete().keys(toDelete).now());
+    return "Done.";
+  }
+}

core/src/test/java/google/registry/batch/SendExpiringCertificateNotificationEmailActionTest.java

@@ -142,7 +142,7 @@ class SendExpiringCertificateNotificationEmailActionTest {
   }
 
   @TestOfyAndSql
-  void sendNotificationEmail_returnsTrue() throws Exception {
+  void sendNotificationEmail_techEMailAsRecipient_returnsTrue() throws Exception {
     X509Certificate expiringCertificate =
         SelfSignedCaCertificate.create(
                 "www.example.tld",
@@ -157,25 +157,64 @@ class SendExpiringCertificateNotificationEmailActionTest {
                 .asBuilder()
                 .setFailoverClientCertificate(cert.get(), clock.nowUtc())
                 .build());
-    ImmutableList<RegistrarContact> contacts =
-        ImmutableList.of(
-            new RegistrarContact.Builder()
-                .setParent(registrar)
-                .setName("Will Doe")
-                .setEmailAddress("will@example-registrar.tld")
-                .setPhoneNumber("+1.3105551213")
-                .setFaxNumber("+1.3105551213")
-                .setTypes(ImmutableSet.of(RegistrarContact.Type.TECH))
-                .setVisibleInWhoisAsAdmin(true)
-                .setVisibleInWhoisAsTech(false)
-                .build());
-    persistSimpleResources(contacts);
-    persistResource(registrar);
+    persistSampleContacts(registrar, Type.TECH);
     assertThat(
             action.sendNotificationEmail(registrar, START_OF_TIME, CertificateType.FAILOVER, cert))
         .isEqualTo(true);
   }
 
+  @TestOfyAndSql
+  void sendNotificationEmail_adminEMailAsRecipient_returnsTrue() throws Exception {
+    X509Certificate expiringCertificate =
+        SelfSignedCaCertificate.create(
+                "www.example.tld",
+                DateTime.parse("2020-09-02T00:00:00Z"),
+                DateTime.parse("2021-06-01T00:00:00Z"))
+            .cert();
+    Optional<String> cert =
+        Optional.of(certificateChecker.serializeCertificate(expiringCertificate));
+    Registrar registrar =
+        persistResource(
+            makeRegistrar1()
+                .asBuilder()
+                .setFailoverClientCertificate(cert.get(), clock.nowUtc())
+                .build());
+    persistSampleContacts(registrar, Type.ADMIN);
+    assertThat(
+            action.sendNotificationEmail(registrar, START_OF_TIME, CertificateType.FAILOVER, cert))
+        .isEqualTo(true);
+  }
+
+  @TestOfyAndSql
+  void sendNotificationEmail_returnsFalse_unsupportedEmailType() throws Exception {
+    Registrar registrar =
+        persistResource(
+            createRegistrar(
+                    "testId",
+                    "testName",
+                    SelfSignedCaCertificate.create(
+                            "www.example.tld",
+                            DateTime.parse("2020-09-02T00:00:00Z"),
+                            DateTime.parse("2021-06-01T00:00:00Z"))
+                        .cert(),
+                    null)
+                .build());
+    persistSampleContacts(registrar, Type.LEGAL);
+    assertThat(
+            action.sendNotificationEmail(
+                registrar,
+                START_OF_TIME,
+                CertificateType.FAILOVER,
+                Optional.of(
+                    certificateChecker.serializeCertificate(
+                        SelfSignedCaCertificate.create(
+                                "www.example.tld",
+                                DateTime.parse("2020-09-02T00:00:00Z"),
+                                DateTime.parse("2021-06-01T00:00:00Z"))
+                            .cert()))))
+        .isEqualTo(false);
+  }
+
   @TestOfyAndSql
   void sendNotificationEmail_returnsFalse_noEmailRecipients() throws Exception {
     X509Certificate expiringCertificate =
@@ -247,93 +286,82 @@ class SendExpiringCertificateNotificationEmailActionTest {
   }
 
   @TestOfyAndSql
-  void sendNotificationEmails_allEmailsBeingAttemptedToSend() throws Exception {
-    X509Certificate expiringCertificate =
-        SelfSignedCaCertificate.create(
-                "www.example.tld",
-                DateTime.parse("2020-09-02T00:00:00Z"),
-                DateTime.parse("2021-06-01T00:00:00Z"))
-            .cert();
-    X509Certificate certificate =
-        SelfSignedCaCertificate.create(
-                "www.example.tld",
-                DateTime.parse("2020-09-02T00:00:00Z"),
-                DateTime.parse("2021-10-01T00:00:00Z"))
-            .cert();
-    int numOfRegistrars = 10;
-    int numOfRegistrarsWithExpiringCertificates = 2;
-    for (int i = 1; i <= numOfRegistrarsWithExpiringCertificates; i++) {
-      persistResource(
-          createRegistrar("oldcert" + i, "name" + i, expiringCertificate, null).build());
+  void sendNotificationEmails_allEmailsBeingSent_onlyMainCertificates() throws Exception {
+    for (int i = 1; i <= 10; i++) {
+      Registrar registrar =
+          persistResource(
+              createRegistrar(
+                      "oldcert" + i,
+                      "name" + i,
+                      SelfSignedCaCertificate.create(
+                              "www.example.tld",
+                              DateTime.parse("2020-09-02T00:00:00Z"),
+                              DateTime.parse("2021-06-01T00:00:00Z"))
+                          .cert(),
+                      null)
+                  .build());
+      persistSampleContacts(registrar, Type.TECH);
     }
-    for (int i = numOfRegistrarsWithExpiringCertificates; i <= numOfRegistrars; i++) {
-      persistResource(createRegistrar("goodcert" + i, "name" + i, certificate, null).build());
-    }
-    assertThat(action.sendNotificationEmails()).isEqualTo(numOfRegistrarsWithExpiringCertificates);
+    assertThat(action.sendNotificationEmails()).isEqualTo(10);
   }
 
   @TestOfyAndSql
-  void sendNotificationEmails_allEmailsBeingAttemptedToSend_onlyMainCertificates()
-      throws Exception {
-    X509Certificate expiringCertificate =
-        SelfSignedCaCertificate.create(
-                "www.example.tld",
-                DateTime.parse("2020-09-02T00:00:00Z"),
-                DateTime.parse("2021-06-01T00:00:00Z"))
-            .cert();
-    int numOfRegistrars = 10;
-    for (int i = 1; i <= numOfRegistrars; i++) {
-      persistResource(
-          createRegistrar("oldcert" + i, "name" + i, expiringCertificate, null).build());
+  void sendNotificationEmails_allEmailsBeingSent_onlyFailOverCertificates() throws Exception {
+    for (int i = 1; i <= 10; i++) {
+      Registrar registrar =
+          persistResource(
+              createRegistrar(
+                      "oldcert" + i,
+                      "name" + i,
+                      null,
+                      SelfSignedCaCertificate.create(
+                              "www.example.tld",
+                              DateTime.parse("2020-09-02T00:00:00Z"),
+                              DateTime.parse("2021-06-01T00:00:00Z"))
+                          .cert())
+                  .build());
+      persistSampleContacts(registrar, Type.TECH);
     }
-    assertThat(action.sendNotificationEmails()).isEqualTo(numOfRegistrars);
+    assertThat(action.sendNotificationEmails()).isEqualTo(10);
   }
 
   @TestOfyAndSql
-  void sendNotificationEmails_allEmailsBeingAttemptedToSend_onlyFailOverCertificates()
-      throws Exception {
+  void sendNotificationEmails_allEmailsBeingSent_mixedOfCertificates() throws Exception {
     X509Certificate expiringCertificate =
         SelfSignedCaCertificate.create(
                 "www.example.tld",
                 DateTime.parse("2020-09-02T00:00:00Z"),
                 DateTime.parse("2021-06-01T00:00:00Z"))
             .cert();
-    int numOfRegistrars = 10;
-    for (int i = 1; i <= numOfRegistrars; i++) {
-      persistResource(
-          createRegistrar("oldcert" + i, "name" + i, null, expiringCertificate).build());
-    }
-    assertThat(action.sendNotificationEmails()).isEqualTo(numOfRegistrars);
-  }
 
-  @TestOfyAndSql
-  void sendNotificationEmails_allEmailsBeingAttemptedToSend_mixedOfCertificates() throws Exception {
-    X509Certificate expiringCertificate =
-        SelfSignedCaCertificate.create(
-                "www.example.tld",
-                DateTime.parse("2020-09-02T00:00:00Z"),
-                DateTime.parse("2021-06-01T00:00:00Z"))
-            .cert();
-    int numOfRegistrars = 10;
-    int numOfExpiringFailOverOnly = 2;
-    int numOfExpiringPrimaryOnly = 3;
-    for (int i = 1; i <= numOfExpiringFailOverOnly; i++) {
-      persistResource(
-          createRegistrar("cl" + i, "expiringFailOverOnly" + i, null, expiringCertificate).build());
+    for (int i = 1; i <= 3; i++) {
+      Registrar registrar =
+          persistResource(
+              createRegistrar(
+                      "cl" + i, "regWIthexpiringFailOverOnly" + i, null, expiringCertificate)
+                  .build());
+      persistSampleContacts(registrar, Type.TECH);
     }
-    for (int i = 1; i <= numOfExpiringPrimaryOnly; i++) {
-      persistResource(
-          createRegistrar("cli" + i, "expiringPrimaryOnly" + i, expiringCertificate, null).build());
+    for (int i = 1; i <= 5; i++) {
+      Registrar registrar =
+          persistResource(
+              createRegistrar(
+                      "cli" + i, "regWithexpiringPrimaryOnly" + i, expiringCertificate, null)
+                  .build());
+      persistSampleContacts(registrar, Type.TECH);
     }
-    for (int i = numOfExpiringFailOverOnly + numOfExpiringPrimaryOnly + 1;
-        i <= numOfRegistrars;
-        i++) {
-      persistResource(
-          createRegistrar("client" + i, "regularReg" + i, expiringCertificate, expiringCertificate)
-              .build());
+    for (int i = 1; i <= 4; i++) {
+      Registrar registrar =
+          persistResource(
+              createRegistrar(
+                      "client" + i,
+                      "regWithTwoExpiring" + i,
+                      expiringCertificate,
+                      expiringCertificate)
+                  .build());
+      persistSampleContacts(registrar, Type.ADMIN);
     }
-    assertThat(action.sendNotificationEmails())
-        .isEqualTo(numOfRegistrars + numOfExpiringFailOverOnly + numOfExpiringPrimaryOnly);
+    assertThat(action.sendNotificationEmails()).isEqualTo(16);
   }
 
   @TestOfyAndSql
@@ -649,17 +677,19 @@ class SendExpiringCertificateNotificationEmailActionTest {
   @TestOfyAndSql
   void run_sentEmails_responseStatusIs200() throws Exception {
     for (int i = 1; i <= 5; i++) {
-      persistResource(
-          createRegistrar(
-                  "id_" + i,
-                  "name" + i,
-                  SelfSignedCaCertificate.create(
-                          "www.example.tld",
-                          DateTime.parse("2020-09-02T00:00:00Z"),
-                          DateTime.parse("2021-06-01T00:00:00Z"))
-                      .cert(),
-                  null)
-              .build());
+      Registrar registrar =
+          persistResource(
+              createRegistrar(
+                      "id_" + i,
+                      "name" + i,
+                      SelfSignedCaCertificate.create(
+                              "www.example.tld",
+                              DateTime.parse("2020-09-02T00:00:00Z"),
+                              DateTime.parse("2021-06-01T00:00:00Z"))
+                          .cert(),
+                      null)
+                  .build());
+      persistSampleContacts(registrar, Type.TECH);
     }
     action.run();
     assertThat(response.getStatus()).isEqualTo(SC_OK);
@@ -667,7 +697,9 @@ class SendExpiringCertificateNotificationEmailActionTest {
         .isEqualTo("Done. Sent 5 expiring certificate notification emails in total.");
   }
 
-  /** Returns a sample registrar with a customized registrar name, client id and certificate* */
+  /**
+   * Returns a sample registrar builder with a customized registrar name, client id and certificate.
+   */
   private Registrar.Builder createRegistrar(
       String registrarId,
       String registrarName,
@@ -706,4 +738,27 @@ class SendExpiringCertificateNotificationEmailActionTest {
     }
     return builder;
   }
+
+  /** Returns persisted sample contacts with a customized contact email type. */
+  private ImmutableList<RegistrarContact> persistSampleContacts(
+      Registrar registrar, RegistrarContact.Type emailType) {
+    return persistSimpleResources(
+        ImmutableList.of(
+            new RegistrarContact.Builder()
+                .setParent(registrar)
+                .setName("Will Doe")
+                .setEmailAddress("will@example-registrar.tld")
+                .setPhoneNumber("+1.0105551213")
+                .setFaxNumber("+1.0105551213")
+                .setTypes(ImmutableSet.of(emailType))
+                .build(),
+            new RegistrarContact.Builder()
+                .setParent(registrar)
+                .setName("Will Smith")
+                .setEmailAddress("will@test-registrar.tld")
+                .setPhoneNumber("+1.3105551213")
+                .setFaxNumber("+1.3105551213")
+                .setTypes(ImmutableSet.of(emailType))
+                .build()));
+  }
 }

core/src/test/java/google/registry/beam/common/RegistryPipelineOptionsTest.java

@@ -19,6 +19,7 @@ import static google.registry.beam.common.RegistryPipelineOptions.validateRegist
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import google.registry.config.RegistryEnvironment;
+import google.registry.persistence.PersistenceModule.JpaTransactionManagerType;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.testing.SystemPropertyExtension;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
@@ -123,4 +124,37 @@ class RegistryPipelineOptionsTest {
     validateRegistryPipelineOptions(options);
     assertThat(options.getProject()).isEqualTo("some-project");
   }
+
+  @Test
+  void jpaTransactionManagerType_default() {
+    RegistryPipelineOptions options =
+        PipelineOptionsFactory.fromArgs(
+                "--registryEnvironment=" + RegistryEnvironment.UNITTEST.name())
+            .withValidation()
+            .as(RegistryPipelineOptions.class);
+    assertThat(options.getJpaTransactionManagerType()).isEqualTo(JpaTransactionManagerType.REGULAR);
+  }
+
+  @Test
+  void jpaTransactionManagerType_regularJpa() {
+    RegistryPipelineOptions options =
+        PipelineOptionsFactory.fromArgs(
+                "--registryEnvironment=" + RegistryEnvironment.UNITTEST.name(),
+                "--jpaTransactionManagerType=REGULAR")
+            .withValidation()
+            .as(RegistryPipelineOptions.class);
+    assertThat(options.getJpaTransactionManagerType()).isEqualTo(JpaTransactionManagerType.REGULAR);
+  }
+
+  @Test
+  void jpaTransactionManagerType_bulkQueryJpa() {
+    RegistryPipelineOptions options =
+        PipelineOptionsFactory.fromArgs(
+                "--registryEnvironment=" + RegistryEnvironment.UNITTEST.name(),
+                "--jpaTransactionManagerType=BULK_QUERY")
+            .withValidation()
+            .as(RegistryPipelineOptions.class);
+    assertThat(options.getJpaTransactionManagerType())
+        .isEqualTo(JpaTransactionManagerType.BULK_QUERY);
+  }
 }

core/src/test/java/google/registry/model/ImmutableObjectSubject.java

@@ -412,6 +412,10 @@ public final class ImmutableObjectSubject extends Subject {
     // don't use ImmutableMap or a stream->collect model since we can have nulls
     Map<Field, Object> result = new LinkedHashMap<>();
     for (Map.Entry<Field, Object> entry : originalFields.entrySet()) {
+      // TODO(b/203685960): filter by @DoNotCompare instead.
+      if (entry.getKey().isAnnotationPresent(ImmutableObject.Insignificant.class)) {
+        continue;
+      }
       if (!ignoredFieldSet.contains(entry.getKey().getName())) {
         result.put(entry.getKey(), entry.getValue());
       }
@@ -426,7 +430,9 @@ public final class ImmutableObjectSubject extends Subject {
     // don't use ImmutableMap or a stream->collect model since we can have nulls
     Map<Field, Object> result = new LinkedHashMap<>();
     for (Map.Entry<Field, Object> entry : originalFields.entrySet()) {
-      if (!entry.getKey().isAnnotationPresent(annotation)) {
+      // TODO(b/203685960): filter by @DoNotCompare instead.
+      if (!entry.getKey().isAnnotationPresent(annotation)
+          && !entry.getKey().isAnnotationPresent(ImmutableObject.Insignificant.class)) {
 
         // Perform any necessary substitutions.
         if (entry.getKey().isAnnotationPresent(ImmutableObject.EmptySetToNull.class)

core/src/test/java/google/registry/model/bulkquery/BulkQueryHelper.java

@@ -0,0 +1,87 @@
+// 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.model.bulkquery;
+
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+
+import google.registry.model.domain.DomainBase;
+import google.registry.model.domain.DomainHistory;
+import google.registry.model.domain.DomainHistory.DomainHistoryId;
+import google.registry.model.domain.GracePeriod;
+import google.registry.model.domain.GracePeriod.GracePeriodHistory;
+import google.registry.model.domain.secdns.DelegationSignerData;
+import google.registry.model.domain.secdns.DomainDsDataHistory;
+import google.registry.model.reporting.DomainTransactionRecord;
+import google.registry.persistence.VKey;
+
+/**
+ * Helpers for bulk-loading {@link google.registry.model.domain.DomainBase} and {@link
+ * google.registry.model.domain.DomainHistory} entities in <em>tests</em>.
+ */
+public class BulkQueryHelper {
+
+  static DomainBase loadAndAssembleDomainBase(String domainRepoId) {
+    return jpaTm()
+        .transact(
+            () ->
+                BulkQueryEntities.assembleDomainBase(
+                    jpaTm().loadByKey(DomainBaseLite.createVKey(domainRepoId)),
+                    jpaTm()
+                        .loadAllOfStream(GracePeriod.class)
+                        .filter(gracePeriod -> gracePeriod.getDomainRepoId().equals(domainRepoId))
+                        .collect(toImmutableSet()),
+                    jpaTm()
+                        .loadAllOfStream(DelegationSignerData.class)
+                        .filter(dsData -> dsData.getDomainRepoId().equals(domainRepoId))
+                        .collect(toImmutableSet()),
+                    jpaTm()
+                        .loadAllOfStream(DomainHost.class)
+                        .filter(domainHost -> domainHost.getDomainRepoId().equals(domainRepoId))
+                        .map(DomainHost::getHostVKey)
+                        .collect(toImmutableSet())));
+  }
+
+  static DomainHistory loadAndAssembleDomainHistory(DomainHistoryId domainHistoryId) {
+    return jpaTm()
+        .transact(
+            () ->
+                BulkQueryEntities.assembleDomainHistory(
+                    jpaTm().loadByKey(VKey.createSql(DomainHistoryLite.class, domainHistoryId)),
+                    jpaTm()
+                        .loadAllOfStream(DomainDsDataHistory.class)
+                        .filter(
+                            domainDsDataHistory ->
+                                domainDsDataHistory.getDomainHistoryId().equals(domainHistoryId))
+                        .collect(toImmutableSet()),
+                    jpaTm()
+                        .loadAllOfStream(DomainHistoryHost.class)
+                        .filter(
+                            domainHistoryHost ->
+                                domainHistoryHost.getDomainHistoryId().equals(domainHistoryId))
+                        .map(DomainHistoryHost::getHostVKey)
+                        .collect(toImmutableSet()),
+                    jpaTm()
+                        .loadAllOfStream(GracePeriodHistory.class)
+                        .filter(
+                            gracePeriodHistory ->
+                                gracePeriodHistory.getDomainHistoryId().equals(domainHistoryId))
+                        .collect(toImmutableSet()),
+                    jpaTm()
+                        .loadAllOfStream(DomainTransactionRecord.class)
+                        .filter(x -> true)
+                        .collect(toImmutableSet())));
+  }
+}

core/src/test/java/google/registry/model/bulkquery/DomainBaseLiteTest.java

@@ -0,0 +1,117 @@
+// 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.model.bulkquery;
+
+import static com.google.common.truth.Truth.assertThat;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static org.joda.time.DateTimeZone.UTC;
+
+import com.google.common.collect.Sets;
+import com.google.common.collect.Sets.SetView;
+import com.google.common.truth.Truth8;
+import google.registry.model.domain.DomainBase;
+import google.registry.testing.AppEngineExtension;
+import google.registry.testing.FakeClock;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.persistence.metamodel.Attribute;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** Unit tests for reading {@link DomainBaseLite}. */
+class DomainBaseLiteTest {
+
+  protected FakeClock fakeClock = new FakeClock(DateTime.now(UTC));
+
+  @RegisterExtension
+  public final AppEngineExtension appEngine =
+      AppEngineExtension.builder().withDatastoreAndCloudSql().withClock(fakeClock).build();
+
+  private final TestSetupHelper setupHelper = new TestSetupHelper(fakeClock);
+
+  @BeforeEach
+  void setUp() {
+    setupHelper.initializeAllEntities();
+  }
+
+  @AfterEach
+  void afterEach() {
+    setupHelper.tearDownBulkQueryJpaTm();
+  }
+
+  @Test
+  void readDomainHost() {
+    setupHelper.applyChangeToDomainAndHistory();
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    Truth8.assertThat(
+            jpaTm().transact(() -> jpaTm().loadAllOf(DomainHost.class)).stream()
+                .map(DomainHost::getHostVKey))
+        .containsExactly(setupHelper.host.createVKey());
+  }
+
+  @Test
+  void domainBaseLiteAttributes_versusDomainBase() {
+    Set<String> domainBaseAttributes =
+        jpaTm()
+            .transact(
+                () ->
+                    jpaTm()
+                        .getEntityManager()
+                        .getMetamodel()
+                        .entity(DomainBase.class)
+                        .getAttributes())
+            .stream()
+            .map(Attribute::getName)
+            .collect(Collectors.toSet());
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    Set<String> domainBaseLiteAttributes =
+        jpaTm()
+            .transact(
+                () ->
+                    jpaTm()
+                        .getEntityManager()
+                        .getMetamodel()
+                        .entity(DomainBaseLite.class)
+                        .getAttributes())
+            .stream()
+            .map(Attribute::getName)
+            .collect(Collectors.toSet());
+
+    assertThat(domainBaseAttributes).containsAtLeastElementsIn(domainBaseLiteAttributes);
+
+    SetView<?> excludedFromDomainBase =
+        Sets.difference(domainBaseAttributes, domainBaseLiteAttributes);
+    assertThat(excludedFromDomainBase)
+        .containsExactly("internalDelegationSignerData", "internalGracePeriods", "nsHosts");
+  }
+
+  @Test
+  void readDomainBaseLite_simple() {
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    assertThat(BulkQueryHelper.loadAndAssembleDomainBase(TestSetupHelper.DOMAIN_REPO_ID))
+        .isEqualTo(setupHelper.domain);
+  }
+
+  @Test
+  void readDomainBaseLite_full() {
+    setupHelper.applyChangeToDomainAndHistory();
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    assertThat(BulkQueryHelper.loadAndAssembleDomainBase(TestSetupHelper.DOMAIN_REPO_ID))
+        .isEqualTo(setupHelper.domain);
+  }
+}

core/src/test/java/google/registry/model/bulkquery/DomainHistoryLiteTest.java

@@ -0,0 +1,125 @@
+// 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.model.bulkquery;
+
+import static com.google.common.truth.Truth.assertThat;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static org.joda.time.DateTimeZone.UTC;
+
+import com.google.common.collect.Sets;
+import com.google.common.collect.Sets.SetView;
+import com.google.common.truth.Truth8;
+import google.registry.model.domain.DomainHistory;
+import google.registry.testing.AppEngineExtension;
+import google.registry.testing.FakeClock;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.persistence.metamodel.Attribute;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** Unit tests for {@link DomainHistoryLite}. */
+public class DomainHistoryLiteTest {
+
+  protected FakeClock fakeClock = new FakeClock(DateTime.now(UTC));
+
+  @RegisterExtension
+  public final AppEngineExtension appEngine =
+      AppEngineExtension.builder().withDatastoreAndCloudSql().withClock(fakeClock).build();
+
+  private final TestSetupHelper setupHelper = new TestSetupHelper(fakeClock);
+
+  @BeforeEach
+  void setUp() {
+    setupHelper.initializeAllEntities();
+  }
+
+  @AfterEach
+  void afterEach() {
+    setupHelper.tearDownBulkQueryJpaTm();
+  }
+
+  @Test
+  void readDomainHistoryHost() {
+    setupHelper.applyChangeToDomainAndHistory();
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    Truth8.assertThat(
+            jpaTm().transact(() -> jpaTm().loadAllOf(DomainHistoryHost.class)).stream()
+                .map(DomainHistoryHost::getHostVKey))
+        .containsExactly(setupHelper.host.createVKey());
+  }
+
+  @Test
+  void domainHistoryLiteAttributes_versusDomainHistory() {
+    Set<String> domainHistoryAttributes =
+        jpaTm()
+            .transact(
+                () ->
+                    jpaTm()
+                        .getEntityManager()
+                        .getMetamodel()
+                        .entity(DomainHistory.class)
+                        .getAttributes())
+            .stream()
+            .map(Attribute::getName)
+            .collect(Collectors.toSet());
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    Set<String> domainHistoryLiteAttributes =
+        jpaTm()
+            .transact(
+                () ->
+                    jpaTm()
+                        .getEntityManager()
+                        .getMetamodel()
+                        .entity(DomainHistoryLite.class)
+                        .getAttributes())
+            .stream()
+            .map(Attribute::getName)
+            .collect(Collectors.toSet());
+
+    assertThat(domainHistoryAttributes).containsAtLeastElementsIn(domainHistoryLiteAttributes);
+
+    SetView<?> excludedFromDomainHistory =
+        Sets.difference(domainHistoryAttributes, domainHistoryLiteAttributes);
+    assertThat(excludedFromDomainHistory)
+        .containsExactly(
+            "dsDataHistories",
+            "gracePeriodHistories",
+            "internalDomainTransactionRecords",
+            "nsHosts");
+  }
+
+  @Test
+  void readDomainHistory_noContent() {
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    assertThat(
+            BulkQueryHelper.loadAndAssembleDomainHistory(
+                setupHelper.domainHistory.getDomainHistoryId()))
+        .isEqualTo(setupHelper.domainHistory);
+  }
+
+  @Test
+  void readDomainHistory_full() {
+    setupHelper.applyChangeToDomainAndHistory();
+    setupHelper.setupBulkQueryJpaTm(appEngine);
+    assertThat(
+            BulkQueryHelper.loadAndAssembleDomainHistory(
+                setupHelper.domainHistory.getDomainHistoryId()))
+        .isEqualTo(setupHelper.domainHistory);
+  }
+}

core/src/test/java/google/registry/model/bulkquery/TestSetupHelper.java

@@ -0,0 +1,210 @@
+// 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.model.bulkquery;
+
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.testing.SqlHelper.saveRegistrar;
+import static google.registry.util.DateTimeUtils.END_OF_TIME;
+import static google.registry.util.DateTimeUtils.START_OF_TIME;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.common.base.Ascii;
+import com.google.common.collect.ImmutableSet;
+import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DesignatedContact;
+import google.registry.model.domain.DomainAuthInfo;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.domain.DomainHistory;
+import google.registry.model.domain.GracePeriod;
+import google.registry.model.domain.Period;
+import google.registry.model.domain.launch.LaunchNotice;
+import google.registry.model.domain.rgp.GracePeriodStatus;
+import google.registry.model.domain.secdns.DelegationSignerData;
+import google.registry.model.eppcommon.AuthInfo.PasswordAuth;
+import google.registry.model.eppcommon.StatusValue;
+import google.registry.model.eppcommon.Trid;
+import google.registry.model.host.HostResource;
+import google.registry.model.registrar.Registrar;
+import google.registry.model.reporting.DomainTransactionRecord;
+import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
+import google.registry.model.reporting.HistoryEntry;
+import google.registry.model.tld.Registry;
+import google.registry.model.transfer.ContactTransferData;
+import google.registry.persistence.BulkQueryJpaFactory;
+import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationTestExtension;
+import google.registry.persistence.transaction.JpaTransactionManager;
+import google.registry.persistence.transaction.TransactionManagerFactory;
+import google.registry.testing.AppEngineExtension;
+import google.registry.testing.DatabaseHelper;
+import google.registry.testing.FakeClock;
+
+/** Entity creation utilities for domain-related tests. */
+class TestSetupHelper {
+
+  public static final String TLD = "tld";
+  public static final String DOMAIN_REPO_ID = "4-TLD";
+  public static final String DOMAIN_NAME = "example.tld";
+  public static final String REGISTRAR_ID = "AnRegistrar";
+
+  private final FakeClock fakeClock;
+
+  Registry registry;
+  Registrar registrar;
+  ContactResource contact;
+  DomainBase domain;
+  DomainHistory domainHistory;
+  HostResource host;
+
+  private JpaTransactionManager originalJpaTm;
+  private JpaTransactionManager bulkQueryJpaTm;
+
+  TestSetupHelper(FakeClock fakeClock) {
+    this.fakeClock = fakeClock;
+  }
+
+  void initializeAllEntities() {
+    registry = putInDb(DatabaseHelper.newRegistry(TLD, Ascii.toUpperCase(TLD)));
+    registrar = saveRegistrar(REGISTRAR_ID);
+    contact = putInDb(createContact(DOMAIN_REPO_ID, REGISTRAR_ID));
+    domain = putInDb(createSimpleDomain(contact));
+    domainHistory = putInDb(createHistoryWithoutContent(domain, fakeClock));
+    host = putInDb(createHost());
+  }
+
+  void applyChangeToDomainAndHistory() {
+    domain = putInDb(createFullDomain(contact, host, fakeClock));
+    domainHistory = putInDb(createFullHistory(domain, fakeClock));
+  }
+
+  void setupBulkQueryJpaTm(AppEngineExtension appEngineExtension) {
+    bulkQueryJpaTm =
+        BulkQueryJpaFactory.createBulkQueryJpaTransactionManager(
+            appEngineExtension
+                .getJpaIntegrationTestExtension()
+                .map(JpaIntegrationTestExtension::getJpaProperties)
+                .orElseThrow(
+                    () -> new IllegalStateException("Expecting JpaIntegrationTestExtension.")),
+            fakeClock);
+    originalJpaTm = TransactionManagerFactory.jpaTm();
+    TransactionManagerFactory.setJpaTm(() -> bulkQueryJpaTm);
+  }
+
+  void tearDownBulkQueryJpaTm() {
+    if (bulkQueryJpaTm != null) {
+      bulkQueryJpaTm.teardown();
+      TransactionManagerFactory.setJpaTm(() -> originalJpaTm);
+    }
+  }
+
+  static ContactResource createContact(String repoId, String registrarId) {
+    return new ContactResource.Builder()
+        .setRepoId(repoId)
+        .setCreationRegistrarId(registrarId)
+        .setTransferData(new ContactTransferData.Builder().build())
+        .setPersistedCurrentSponsorRegistrarId(registrarId)
+        .build();
+  }
+
+  static DomainBase createSimpleDomain(ContactResource contact) {
+    return DatabaseHelper.newDomainBase(DOMAIN_NAME, DOMAIN_REPO_ID, contact)
+        .asBuilder()
+        .setCreationRegistrarId(REGISTRAR_ID)
+        .setPersistedCurrentSponsorRegistrarId(REGISTRAR_ID)
+        .build();
+  }
+
+  static DomainBase createFullDomain(
+      ContactResource contact, HostResource host, FakeClock fakeClock) {
+    return createSimpleDomain(contact)
+        .asBuilder()
+        .setDomainName(DOMAIN_NAME)
+        .setRepoId(DOMAIN_REPO_ID)
+        .setCreationRegistrarId(REGISTRAR_ID)
+        .setLastEppUpdateTime(fakeClock.nowUtc())
+        .setLastEppUpdateRegistrarId(REGISTRAR_ID)
+        .setLastTransferTime(fakeClock.nowUtc())
+        .setNameservers(host.createVKey())
+        .setStatusValues(
+            ImmutableSet.of(
+                StatusValue.CLIENT_DELETE_PROHIBITED,
+                StatusValue.SERVER_DELETE_PROHIBITED,
+                StatusValue.SERVER_TRANSFER_PROHIBITED,
+                StatusValue.SERVER_UPDATE_PROHIBITED,
+                StatusValue.SERVER_RENEW_PROHIBITED,
+                StatusValue.SERVER_HOLD))
+        .setContacts(
+            ImmutableSet.of(
+                DesignatedContact.create(DesignatedContact.Type.ADMIN, contact.createVKey())))
+        .setSubordinateHosts(ImmutableSet.of("ns1.example.com"))
+        .setPersistedCurrentSponsorRegistrarId(REGISTRAR_ID)
+        .setRegistrationExpirationTime(fakeClock.nowUtc().plusYears(1))
+        .setAuthInfo(DomainAuthInfo.create(PasswordAuth.create("password")))
+        .setDsData(ImmutableSet.of(DelegationSignerData.create(1, 2, 3, new byte[] {0, 1, 2})))
+        .setLaunchNotice(LaunchNotice.create("tcnid", "validatorId", START_OF_TIME, START_OF_TIME))
+        .setSmdId("smdid")
+        .addGracePeriod(
+            GracePeriod.create(
+                GracePeriodStatus.ADD, DOMAIN_REPO_ID, END_OF_TIME, REGISTRAR_ID, null, 100L))
+        .build();
+  }
+
+  static HostResource createHost() {
+    return new HostResource.Builder()
+        .setRepoId("host1")
+        .setHostName("ns1.example.com")
+        .setCreationRegistrarId(REGISTRAR_ID)
+        .setPersistedCurrentSponsorRegistrarId(REGISTRAR_ID)
+        .build();
+  }
+
+  static DomainTransactionRecord createDomainTransactionRecord(FakeClock fakeClock) {
+    return new DomainTransactionRecord.Builder()
+        .setTld(TLD)
+        .setReportingTime(fakeClock.nowUtc())
+        .setReportField(TransactionReportField.NET_ADDS_1_YR)
+        .setReportAmount(1)
+        .build();
+  }
+
+  static DomainHistory createHistoryWithoutContent(DomainBase domain, FakeClock fakeClock) {
+    return new DomainHistory.Builder()
+        .setType(HistoryEntry.Type.DOMAIN_CREATE)
+        .setXmlBytes("<xml></xml>".getBytes(UTF_8))
+        .setModificationTime(fakeClock.nowUtc())
+        .setRegistrarId(REGISTRAR_ID)
+        .setTrid(Trid.create("ABC-123", "server-trid"))
+        .setBySuperuser(false)
+        .setReason("reason")
+        .setRequestedByRegistrar(true)
+        .setDomainRepoId(domain.getRepoId())
+        .setOtherRegistrarId("otherClient")
+        .setPeriod(Period.create(1, Period.Unit.YEARS))
+        .build();
+  }
+
+  static DomainHistory createFullHistory(DomainBase domain, FakeClock fakeClock) {
+    return createHistoryWithoutContent(domain, fakeClock)
+        .asBuilder()
+        .setType(HistoryEntry.Type.DOMAIN_TRANSFER_APPROVE)
+        .setDomain(domain)
+        .setDomainTransactionRecords(ImmutableSet.of(createDomainTransactionRecord(fakeClock)))
+        .build();
+  }
+
+  static <T> T putInDb(T entity) {
+    jpaTm().transact(() -> jpaTm().put(entity));
+    return jpaTm().transact(() -> jpaTm().loadByEntity(entity));
+  }
+}

core/src/test/java/google/registry/model/replay/EntityTest.java

@@ -51,8 +51,7 @@ public class EntityTest {
 
   @Test
   void testSqlEntityPersistence() {
-    try (ScanResult scanResult =
-        new ClassGraph().enableAnnotationInfo().whitelistPackages("google.registry").scan()) {
+    try (ScanResult scanResult = scanForClasses()) {
       // All javax.persistence entities must implement SqlEntity and vice versa
       ImmutableSet<String> javaxPersistenceClasses =
           getAllClassesWithAnnotation(scanResult, javax.persistence.Entity.class.getName());
@@ -75,8 +74,7 @@ public class EntityTest {
     // For replication, we need to be able to convert from Key -> VKey for the relevant classes.
     // This means that the relevant classes must have non-composite Objectify keys or must have a
     // createVKey method
-    try (ScanResult scanResult =
-        new ClassGraph().enableAnnotationInfo().whitelistPackages("google.registry").scan()) {
+    try (ScanResult scanResult = scanForClasses()) {
       ImmutableSet<Class<?>> datastoreEntityClasses =
           getClasses(scanResult.getClassesImplementing(DatastoreEntity.class.getName()));
       // some classes aren't converted so they aren't relevant
@@ -126,9 +124,12 @@ public class EntityTest {
     return classInfoList.stream()
         .filter(ClassInfo::isStandardClass)
         .map(ClassInfo::loadClass)
-        .filter(clazz -> !clazz.isAnnotationPresent(EntityForTesting.class))
-        .filter(clazz -> !clazz.isAnnotationPresent(Embed.class))
-        .filter(clazz -> !NON_CONVERTED_CLASSES.contains(clazz))
+        .filter(
+            clazz ->
+                !clazz.isAnnotationPresent(EntityForTesting.class)
+                    && !clazz.isAnnotationPresent(Embed.class)
+                    && !NON_CONVERTED_CLASSES.contains(clazz)
+                    && !clazz.getName().contains("Test"))
         .collect(toImmutableSet());
   }
 
@@ -136,6 +137,14 @@ public class EntityTest {
     return getClasses(classInfoList).stream().map(Class::getName).collect(toImmutableSet());
   }
 
+  private ScanResult scanForClasses() {
+    return new ClassGraph()
+        .enableAnnotationInfo()
+        .ignoreClassVisibility()
+        .acceptPackages("google.registry")
+        .scan();
+  }
+
   /** Entities that are solely used for testing, to avoid scanning them in {@link EntityTest}. */
   @Target(ElementType.TYPE)
   @Retention(RetentionPolicy.RUNTIME)

core/src/test/java/google/registry/model/smd/SignedMarkRevocationListTest.java

@@ -30,7 +30,8 @@ import org.junit.jupiter.api.extension.RegisterExtension;
 public class SignedMarkRevocationListTest {
 
   @RegisterExtension
-  public final AppEngineExtension appEngine = AppEngineExtension.builder().withCloudSql().build();
+  public final AppEngineExtension appEngine =
+      AppEngineExtension.builder().withDatastoreAndCloudSql().build();
 
   private final FakeClock clock = new FakeClock(DateTime.parse("2013-01-01T00:00:00Z"));
 

core/src/test/java/google/registry/persistence/transaction/JpaTransactionManagerExtension.java

@@ -45,7 +45,6 @@ import java.sql.Driver;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -172,24 +171,39 @@ abstract class JpaTransactionManagerExtension implements BeforeEachCallback, Aft
       exporter.export(getTestEntities(), tempSqlFile);
       executeSql(new String(Files.readAllBytes(tempSqlFile.toPath()), StandardCharsets.UTF_8));
     }
+    assertReasonableNumDbConnections();
+    emf = createEntityManagerFactory(getJpaProperties());
+    emfEntityHash = entityHash;
+  }
 
-    ImmutableMap<String, String> properties = PersistenceModule.provideDefaultDatabaseConfigs();
+  /**
+   * Returns the full set of properties for setting up JPA {@link EntityManagerFactory} to the test
+   * database. This allows creation of customized JPA by individual tests.
+   *
+   * <p>Test that create {@code EntityManagerFactory} instances are reponsible for tearing them
+   * down.
+   */
+  public ImmutableMap<String, String> getJpaProperties() {
+    Map<String, String> mergedProperties =
+        Maps.newHashMap(PersistenceModule.provideDefaultDatabaseConfigs());
     if (!userProperties.isEmpty()) {
-      // If there are user properties, create a new properties object with these added.
-      Map<String, String> mergedProperties = Maps.newHashMap();
-      mergedProperties.putAll(properties);
       mergedProperties.putAll(userProperties);
-      properties = ImmutableMap.copyOf(mergedProperties);
     }
+    mergedProperties.put(Environment.URL, getJdbcUrl());
+    mergedProperties.put(Environment.USER, database.getUsername());
+    mergedProperties.put(Environment.PASS, database.getPassword());
+    // Tell Postgresql JDBC driver to retry on errors caused by out-of-band schema change between
+    // tests while the connection pool stays open (e.g., "cached plan must not change result type").
+    // We don't set this property in production since it has performance impact, and production
+    // schema is always compatible with the binary (enforced by our release process).
+    mergedProperties.put("hibernate.hikari.dataSource.autosave", "conservative");
+
     // Forbid Hibernate push to stay consistent with flyway-based schema management.
     checkState(
-        Objects.equals(properties.get(Environment.HBM2DDL_AUTO), "none"),
+        Objects.equals(mergedProperties.get(Environment.HBM2DDL_AUTO), "none"),
         "The HBM2DDL_AUTO property must be 'none'.");
-    assertReasonableNumDbConnections();
-    emf =
-        createEntityManagerFactory(
-            getJdbcUrl(), database.getUsername(), database.getPassword(), properties);
-    emfEntityHash = entityHash;
+
+    return ImmutableMap.copyOf(mergedProperties);
   }
 
   @Override
@@ -307,15 +321,7 @@ abstract class JpaTransactionManagerExtension implements BeforeEachCallback, Aft
   }
 
   /** Constructs the {@link EntityManagerFactory} instance. */
-  private EntityManagerFactory createEntityManagerFactory(
-      String jdbcUrl, String username, String password, ImmutableMap<String, String> configs) {
-    HashMap<String, String> properties = Maps.newHashMap(configs);
-    properties.put(Environment.URL, jdbcUrl);
-    properties.put(Environment.USER, username);
-    properties.put(Environment.PASS, password);
-    // Tell Postgresql JDBC driver to expect out-of-band schema change.
-    properties.put("hibernate.hikari.dataSource.autosave", "conservative");
-
+  private EntityManagerFactory createEntityManagerFactory(ImmutableMap<String, String> properties) {
     ParsedPersistenceXmlDescriptor descriptor =
         PersistenceXmlUtility.getParsedPersistenceXmlDescriptor();
 

core/src/test/java/google/registry/reporting/icann/IcannReportingUploadActionTest.java

@@ -294,8 +294,9 @@ class IcannReportingUploadActionTest {
         .that(logHandler)
         .hasLogAtLevelWithMessage(
             Level.SEVERE,
-            "Could not upload ICANN_UPLOAD_ACTIVITY report for tld because file"
-                + " tld-activity-200512.csv did not exist");
+            "Could not upload ICANN_UPLOAD_ACTIVITY report for tld because file "
+                + "tld-activity-200512.csv (object icann/monthly/2005-12/tld-activity-200512.csv in"
+                + " bucket basin) did not exist.");
   }
 
   @TestOfyAndSql
@@ -310,9 +311,9 @@ class IcannReportingUploadActionTest {
         .that(logHandler)
         .hasLogAtLevelWithMessage(
             Level.INFO,
-            "Could not upload ICANN_UPLOAD_ACTIVITY report for foo because file"
-                + " foo-activity-200607.csv did not exist. This report may not have been staged"
-                + " yet.");
+            "Could not upload ICANN_UPLOAD_ACTIVITY report for foo because file "
+                + "foo-activity-200607.csv (object icann/monthly/2006-07/foo-activity-200607.csv in"
+                + " bucket basin) did not exist. This report may not have been staged yet.");
   }
 
   @TestOfyAndSql

core/src/test/java/google/registry/testing/AppEngineExtension.java

@@ -146,6 +146,10 @@ public final class AppEngineExtension implements BeforeEachCallback, AfterEachCa
   private ImmutableList<Class<?>> ofyTestEntities;
   private ImmutableList<Class<?>> jpaTestEntities;
 
+  public Optional<JpaIntegrationTestExtension> getJpaIntegrationTestExtension() {
+    return Optional.ofNullable(jpaIntegrationTestExtension);
+  }
+
   /** Builder for {@link AppEngineExtension}. */
   public static class Builder {
 

core/src/test/java/google/registry/tools/javascrap/CreateSyntheticHistoryEntriesPipelineTest.java

@@ -0,0 +1,115 @@
+// 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.tools.javascrap;
+
+import static com.google.common.truth.Truth.assertThat;
+import static google.registry.model.ImmutableObjectSubject.assertAboutImmutableObjects;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.removeTmOverrideForTest;
+import static google.registry.persistence.transaction.TransactionManagerFactory.setTmOverrideForTest;
+import static google.registry.testing.DatabaseHelper.createTld;
+import static google.registry.testing.DatabaseHelper.newDomainBase;
+import static google.registry.testing.DatabaseHelper.persistActiveHost;
+import static google.registry.testing.DatabaseHelper.persistNewRegistrar;
+import static google.registry.testing.DatabaseHelper.persistSimpleResource;
+
+import com.google.common.collect.ImmutableList;
+import google.registry.beam.TestPipelineExtension;
+import google.registry.model.EppResource;
+import google.registry.model.contact.ContactHistory;
+import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.domain.DomainHistory;
+import google.registry.model.host.HostHistory;
+import google.registry.model.host.HostResource;
+import google.registry.model.reporting.HistoryEntry;
+import google.registry.persistence.transaction.JpaTestExtensions;
+import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationTestExtension;
+import google.registry.testing.DatastoreEntityExtension;
+import google.registry.testing.FakeClock;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** Unit tests for {@link CreateSyntheticHistoryEntriesPipeline}. */
+public class CreateSyntheticHistoryEntriesPipelineTest {
+
+  FakeClock clock = new FakeClock();
+
+  @RegisterExtension
+  JpaIntegrationTestExtension jpaEextension =
+      new JpaTestExtensions.Builder().withClock(clock).buildIntegrationTestExtension();
+
+  @RegisterExtension
+  DatastoreEntityExtension datastoreEntityExtension =
+      new DatastoreEntityExtension().allThreads(true);
+
+  @RegisterExtension TestPipelineExtension pipeline = TestPipelineExtension.create();
+
+  DomainBase domain;
+  ContactResource contact;
+  HostResource host;
+
+  @BeforeEach
+  void beforeEach() {
+    setTmOverrideForTest(jpaTm());
+    persistNewRegistrar("TheRegistrar");
+    persistNewRegistrar("NewRegistrar");
+    createTld("tld");
+    host = persistActiveHost("external.com");
+    domain =
+        persistSimpleResource(
+            newDomainBase("example.tld").asBuilder().setNameservers(host.createVKey()).build());
+    contact = jpaTm().transact(() -> jpaTm().loadByKey(domain.getRegistrant()));
+    clock.advanceOneMilli();
+  }
+
+  @AfterEach
+  void afterEach() {
+    removeTmOverrideForTest();
+  }
+
+  @Test
+  void testSuccess() {
+    assertThat(jpaTm().transact(() -> jpaTm().loadAllOf(DomainHistory.class))).isEmpty();
+    assertThat(jpaTm().transact(() -> jpaTm().loadAllOf(ContactHistory.class))).isEmpty();
+    assertThat(jpaTm().transact(() -> jpaTm().loadAllOf(HostHistory.class))).isEmpty();
+    CreateSyntheticHistoryEntriesPipeline.setup(pipeline, "NewRegistrar");
+    pipeline.run().waitUntilFinish();
+    validateHistoryEntry(DomainHistory.class, domain);
+    validateHistoryEntry(ContactHistory.class, contact);
+    validateHistoryEntry(HostHistory.class, host);
+  }
+
+  private static <T extends EppResource> void validateHistoryEntry(
+      Class<? extends HistoryEntry> historyClazz, T resource) {
+    ImmutableList<? extends HistoryEntry> historyEntries =
+        jpaTm().transact(() -> jpaTm().loadAllOf(historyClazz));
+    assertThat(historyEntries.size()).isEqualTo(1);
+    HistoryEntry historyEntry = historyEntries.get(0);
+    assertThat(historyEntry.getType()).isEqualTo(HistoryEntry.Type.SYNTHETIC);
+    assertThat(historyEntry.getRegistrarId()).isEqualTo("NewRegistrar");
+    EppResource embeddedResource;
+    if (historyEntry instanceof DomainHistory) {
+      embeddedResource = ((DomainHistory) historyEntry).getDomainContent().get();
+    } else if (historyEntry instanceof ContactHistory) {
+      embeddedResource = ((ContactHistory) historyEntry).getContactBase().get();
+    } else {
+      embeddedResource = ((HostHistory) historyEntry).getHostBase().get();
+    }
+    assertAboutImmutableObjects().that(embeddedResource).hasFieldsEqualTo(resource);
+  }
+}

db/src/main/resources/sql/schema/db-schema.sql.generated

@@ -412,12 +412,12 @@
 
     create table "DomainTransactionRecord" (
        id  bigserial not null,
+        domain_repo_id text,
+        history_revision_id int8,
         report_amount int4 not null,
         report_field text not null,
         reporting_time timestamptz not null,
         tld text not null,
-        domain_repo_id text,
-        history_revision_id int8,
         primary key (id)
     );
 

java-format/google-java-format-diff.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python2.7
 #
 #===- google-java-format-diff.py - google-java-format Diff Reformatter -----===#
 #
@@ -24,7 +23,6 @@ For perforce users:
 import argparse
 import difflib
 import re
-import string
 import subprocess
 import io
 import sys
@@ -99,7 +97,7 @@ def main():
     base_command = [binary]
 
   # Reformat files containing changes in place.
-  for filename, lines in lines_by_file.iteritems():
+  for filename, lines in lines_by_file.items():
     if args.i and args.verbose:
       print('Formatting ' + filename)
     command = base_command[:]
@@ -120,11 +118,11 @@ def main():
     if not args.i:
       with open(filename) as f:
         code = f.readlines()
-      formatted_code = io.BytesIO(stdout).readlines()
+      formatted_code = io.StringIO(stdout.decode()).readlines()
       diff = difflib.unified_diff(code, formatted_code,
                                   filename, filename,
                                   '(before formatting)', '(after formatting)')
-      diff_string = string.join(diff, '')
+      diff_string = ''.join(diff)
       if len(diff_string) > 0:
         sys.stdout.write(diff_string)
 

java-format/google-java-format-git-diff.sh

@@ -42,6 +42,16 @@ where:
 SCRIPT_DIR="$(realpath $(dirname $0))"
 JAR_NAME="google-java-format-1.8-all-deps.jar"
 
+# Make sure we have a valid python interpreter.
+if [ -z "$PYTHON" ]; then
+  echo "You must specify the name of a python3 interpreter in the PYTHON" \
+       "environment variable."
+  exit 1
+elif ! "$PYTHON" -c ''; then
+  echo "Invalid python interpreter: $PYTHON"
+  exit 1
+fi
+
 # Locate the java binary.
 if [ -n "$JAVA_HOME" ]; then
   JAVA_BIN="$JAVA_HOME/bin/java"
@@ -69,10 +79,14 @@ function runGoogleJavaFormatAgainstDiffs() {
   shift
 
   git diff -U0 "$forkPoint" | \
-      ${SCRIPT_DIR}/google-java-format-diff.py \
-      --java-binary "$JAVA_BIN" \
-      --google-java-format-jar "${SCRIPT_DIR}/${JAR_NAME}" \
-      -p1 "$@" | tee gjf.out
+      "${PYTHON}" "${SCRIPT_DIR}/google-java-format-diff.py" \
+          --java-binary "$JAVA_BIN" \
+          --google-java-format-jar "${SCRIPT_DIR}/${JAR_NAME}" \
+          -p1 "$@" | \
+      tee gjf.out
+
+  # If any of the commands in the last pipe failed, return false.
+  [[ ! "${PIPESTATUS[@]}" =~ [^0\ ] ]]
 }
 
 # Show the file names in a diff preceeded by a message.
@@ -96,7 +110,11 @@ function callGoogleJavaFormatDiff() {
   local callResult
   case "$1" in
     "check")
-      local output=$(runGoogleJavaFormatAgainstDiffs "$forkPoint")
+      # We need to do explicit checks for an error and "exit 1" if there was
+      # one here (though not elsewhere), "set -e" doesn't catch this case,
+      # it's not clear why.
+      local output
+      output=$(runGoogleJavaFormatAgainstDiffs "$forkPoint") || exit 1
       echo "$output" | showFileNames "\033[1mNeeds formatting: "
       callResult=$(echo -n "$output" | wc -l)
       ;;