Make transaction isolation level the first argument to transact() (#2329)

This makes the callsites look neater, as the work to execute itself is often a
many line lambda, whereas the transaction isolation level is not more than a
couple dozen characters.

.java-version

@@ -0,0 +1 @@
+17

build.gradle

@@ -383,8 +383,9 @@ subprojects {
   apply from: "${rootDir.path}/java_common.gradle"
 
   // When changing Java version here, be sure to update BEAM Java runtime:
-  // in core/build.gradle, search for `flex-template-base-image` and update
-  // the parameter value.
+  // search for `flex-template-base-image` and update the parameter value.
+  // There are at least two instances, one in core/build.gradle, one in
+  // release/stage_beam_pipeline.sh
   sourceCompatibility = JavaVersion.VERSION_17
   targetCompatibility = JavaVersion.VERSION_17
 

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

@@ -207,7 +207,7 @@ public class CloudTasksUtils implements Serializable {
       Service service,
       Multimap<String, String> params,
       Optional<Integer> jitterSeconds) {
-    if (!jitterSeconds.isPresent() || jitterSeconds.get() <= 0) {
+    if (jitterSeconds.isEmpty() || jitterSeconds.get() <= 0) {
       return createTask(path, method, service, params);
     }
     return createTaskWithDelay(

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

@@ -171,7 +171,7 @@ public class DeleteExpiredDomainsAction implements Runnable {
         tm().transact(
                 () -> {
                   Domain transDomain = tm().loadByKey(domain.createVKey());
-                  if (!domain.getAutorenewEndTime().isPresent()
+                  if (domain.getAutorenewEndTime().isEmpty()
                       || domain.getAutorenewEndTime().get().isAfter(tm().getTransactionTime())) {
                     logger.atSevere().log(
                         "Failed to delete domain %s because of its autorenew end time: %s.",

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

@@ -151,7 +151,7 @@ public class SendExpiringCertificateNotificationEmailAction implements Runnable
       DateTime lastExpiringCertNotificationSentDate,
       CertificateType certificateType,
       Optional<String> certificate) {
-    if (!certificate.isPresent()
+    if (certificate.isEmpty()
         || !certificateChecker.shouldReceiveExpiringNotification(
             lastExpiringCertNotificationSentDate, certificate.get())) {
       return false;

core/src/main/java/google/registry/beam/billing/BillingEvent.java

@@ -21,26 +21,21 @@ import google.registry.reporting.billing.BillingModule;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.io.Serializable;
 import java.util.regex.Pattern;
 import org.apache.beam.sdk.coders.AtomicCoder;
 import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.DoubleCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
 import org.joda.time.DateTime;
 import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
 
-/**
- * A POJO representing a single billable event, parsed from a {@code SchemaAndRecord}.
- *
- * <p>This is a trivially serializable class that allows Beam to transform the results of a Cloud
- * SQL query into a standard Java representation, giving us the type guarantees and ease of
- * manipulation Cloud SQL lacks.
- */
+/** A POJO representing a single billable event, parsed from a {@code SchemaAndRecord}. */
 @AutoValue
-public abstract class BillingEvent implements Serializable {
-
-  private static final long serialVersionUID = -3593088371541450077L;
+public abstract class BillingEvent {
 
   private static final DateTimeFormatter DATE_TIME_FORMATTER =
       DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss zzz");
@@ -85,7 +80,7 @@ public abstract class BillingEvent implements Serializable {
   /** Returns the tld this event was generated for. */
   abstract String tld();
 
-  /** Returns the billable action this event was generated for (i.e. RENEW, CREATE, TRANSFER...) */
+  /** Returns the billable action this event was generated for (i.e., RENEW, CREATE, TRANSFER...) */
   abstract String action();
 
   /** Returns the fully qualified domain name this event was generated for. */
@@ -97,7 +92,7 @@ public abstract class BillingEvent implements Serializable {
   /** Returns the number of years this billing event is made out for. */
   abstract int years();
 
-  /** Returns the 3-letter currency code for the billing event (i.e. USD or JPY.) */
+  /** Returns the 3-letter currency code for the billing event (i.e., USD or JPY.) */
   abstract String currency();
 
   /** Returns the cost associated with this billing event. */
@@ -203,9 +198,7 @@ public abstract class BillingEvent implements Serializable {
 
   /** Key for each {@code BillingEvent}, when aggregating for the overall invoice. */
   @AutoValue
-  abstract static class InvoiceGroupingKey implements Serializable {
-
-    private static final long serialVersionUID = -151561764235256205L;
+  abstract static class InvoiceGroupingKey {
 
     private static final ImmutableList<String> INVOICE_HEADERS =
         ImmutableList.of(
@@ -277,8 +270,14 @@ public abstract class BillingEvent implements Serializable {
 
     /** Coder that provides deterministic (de)serialization for {@code InvoiceGroupingKey}. */
     static class InvoiceGroupingKeyCoder extends AtomicCoder<InvoiceGroupingKey> {
+      private static final Coder<String> stringCoder = StringUtf8Coder.of();
+      private static final InvoiceGroupingKeyCoder INSTANCE = new InvoiceGroupingKeyCoder();
 
-      private static final long serialVersionUID = 6680701524304107547L;
+      public static InvoiceGroupingKeyCoder of() {
+        return INSTANCE;
+      }
+
+      private InvoiceGroupingKeyCoder() {}
 
       @Override
       public void encode(InvoiceGroupingKey value, OutputStream outStream) throws IOException {
@@ -295,7 +294,6 @@ public abstract class BillingEvent implements Serializable {
 
       @Override
       public InvoiceGroupingKey decode(InputStream inStream) throws IOException {
-        Coder<String> stringCoder = StringUtf8Coder.of();
         return new AutoValue_BillingEvent_InvoiceGroupingKey(
             stringCoder.decode(inStream),
             stringCoder.decode(inStream),
@@ -308,4 +306,55 @@ public abstract class BillingEvent implements Serializable {
       }
     }
   }
+
+  static class BillingEventCoder extends AtomicCoder<BillingEvent> {
+    private static final Coder<String> stringCoder = StringUtf8Coder.of();
+    private static final Coder<Integer> integerCoder = VarIntCoder.of();
+    private static final Coder<Long> longCoder = VarLongCoder.of();
+    private static final Coder<Double> doubleCoder = DoubleCoder.of();
+    private static final BillingEventCoder INSTANCE = new BillingEventCoder();
+
+    static NullableCoder<BillingEvent> ofNullable() {
+      return NullableCoder.of(INSTANCE);
+    }
+
+    private BillingEventCoder() {}
+
+    @Override
+    public void encode(BillingEvent value, OutputStream outStream) throws IOException {
+      longCoder.encode(value.id(), outStream);
+      stringCoder.encode(DATE_TIME_FORMATTER.print(value.billingTime()), outStream);
+      stringCoder.encode(DATE_TIME_FORMATTER.print(value.eventTime()), outStream);
+      stringCoder.encode(value.registrarId(), outStream);
+      stringCoder.encode(value.billingId(), outStream);
+      stringCoder.encode(value.poNumber(), outStream);
+      stringCoder.encode(value.tld(), outStream);
+      stringCoder.encode(value.action(), outStream);
+      stringCoder.encode(value.domain(), outStream);
+      stringCoder.encode(value.repositoryId(), outStream);
+      integerCoder.encode(value.years(), outStream);
+      stringCoder.encode(value.currency(), outStream);
+      doubleCoder.encode(value.amount(), outStream);
+      stringCoder.encode(value.flags(), outStream);
+    }
+
+    @Override
+    public BillingEvent decode(InputStream inStream) throws IOException {
+      return new AutoValue_BillingEvent(
+          longCoder.decode(inStream),
+          DATE_TIME_FORMATTER.parseDateTime(stringCoder.decode(inStream)),
+          DATE_TIME_FORMATTER.parseDateTime(stringCoder.decode(inStream)),
+          stringCoder.decode(inStream),
+          stringCoder.decode(inStream),
+          stringCoder.decode(inStream),
+          stringCoder.decode(inStream),
+          stringCoder.decode(inStream),
+          stringCoder.decode(inStream),
+          stringCoder.decode(inStream),
+          integerCoder.decode(inStream),
+          stringCoder.decode(inStream),
+          doubleCoder.decode(inStream),
+          stringCoder.decode(inStream));
+    }
+  }
 }

core/src/main/java/google/registry/beam/billing/InvoicingPipeline.java

@@ -18,6 +18,7 @@ import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static org.apache.beam.sdk.values.TypeDescriptors.strings;
 
 import com.google.common.flogger.FluentLogger;
+import google.registry.beam.billing.BillingEvent.BillingEventCoder;
 import google.registry.beam.billing.BillingEvent.InvoiceGroupingKey;
 import google.registry.beam.billing.BillingEvent.InvoiceGroupingKey.InvoiceGroupingKeyCoder;
 import google.registry.beam.common.RegistryJpaIO;
@@ -30,6 +31,7 @@ import google.registry.reporting.billing.BillingModule;
 import google.registry.util.DomainNameUtils;
 import google.registry.util.ResourceUtils;
 import google.registry.util.SqlTemplate;
+import java.io.Serial;
 import java.io.Serializable;
 import java.time.YearMonth;
 import java.util.Objects;
@@ -37,13 +39,13 @@ import java.util.Optional;
 import java.util.regex.Pattern;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
-import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.io.FileIO;
 import org.apache.beam.sdk.io.TextIO;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.transforms.Contextful;
 import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.Distinct;
 import org.apache.beam.sdk.transforms.Filter;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -65,7 +67,7 @@ import org.joda.money.CurrencyUnit;
  */
 public class InvoicingPipeline implements Serializable {
 
-  private static final long serialVersionUID = 5386330443625580081L;
+  @Serial private static final long serialVersionUID = 5386330443625580081L;
 
   private static final Pattern SQL_COMMENT_REGEX =
       Pattern.compile("^\\s*--.*\\n", Pattern.MULTILINE);
@@ -97,13 +99,11 @@ public class InvoicingPipeline implements Serializable {
     Read<Object[], google.registry.beam.billing.BillingEvent> read =
         RegistryJpaIO.<Object[], google.registry.beam.billing.BillingEvent>read(
                 makeCloudSqlQuery(options.getYearMonth()), false, row -> parseRow(row).orElse(null))
-            .withCoder(SerializableCoder.of(google.registry.beam.billing.BillingEvent.class));
-
-    PCollection<google.registry.beam.billing.BillingEvent> billingEventsWithNulls =
-        pipeline.apply("Read BillingEvents from Cloud SQL", read);
-
-    // Remove null billing events
-    return billingEventsWithNulls.apply(Filter.by(Objects::nonNull));
+            .withCoder(BillingEventCoder.ofNullable());
+    return pipeline
+        .apply("Read BillingEvents from Cloud SQL", read)
+        .apply("Remove null elements", Filter.by(Objects::nonNull))
+        .apply("Remove duplicates", Distinct.create());
   }
 
   private static Optional<google.registry.beam.billing.BillingEvent> parseRow(Object[] row) {
@@ -142,7 +142,7 @@ public class InvoicingPipeline implements Serializable {
       extends PTransform<
           PCollection<google.registry.beam.billing.BillingEvent>, PCollection<String>> {
 
-    private static final long serialVersionUID = -8090619008258393728L;
+    @Serial private static final long serialVersionUID = -8090619008258393728L;
 
     @Override
     public PCollection<String> expand(
@@ -152,9 +152,9 @@ public class InvoicingPipeline implements Serializable {
               "Map to invoicing key",
               MapElements.into(TypeDescriptor.of(InvoiceGroupingKey.class))
                   .via(google.registry.beam.billing.BillingEvent::getInvoiceGroupingKey))
+          .setCoder(InvoiceGroupingKeyCoder.of())
           .apply(
               "Filter out free events", Filter.by((InvoiceGroupingKey key) -> key.unitPrice() != 0))
-          .setCoder(new InvoiceGroupingKeyCoder())
           .apply("Count occurrences", Count.perElement())
           .apply(
               "Format as CSVs",

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

@@ -219,8 +219,7 @@ public final class RegistryJpaIO {
                 () -> {
                   query.stream().map(resultMapper::apply).forEach(outputReceiver::output);
                   return null;
-                },
-                null);
+                });
       }
     }
   }

core/src/main/java/google/registry/bsa/BsaDownloadAction.java

@@ -116,7 +116,7 @@ public class BsaDownloadAction implements Runnable {
       return null;
     }
     Optional<DownloadSchedule> scheduleOptional = downloadScheduler.schedule();
-    if (!scheduleOptional.isPresent()) {
+    if (scheduleOptional.isEmpty()) {
       logger.atInfo().log("Nothing to do.");
       return null;
     }

core/src/main/java/google/registry/bsa/BsaRefreshAction.java

@@ -102,7 +102,7 @@ public class BsaRefreshAction implements Runnable {
       return null;
     }
     Optional<RefreshSchedule> maybeSchedule = scheduler.schedule();
-    if (!maybeSchedule.isPresent()) {
+    if (maybeSchedule.isEmpty()) {
       logger.atInfo().log("No completed downloads yet. Exiting.");
       return null;
     }

core/src/main/java/google/registry/bsa/BsaTransactions.java

@@ -37,12 +37,12 @@ public final class BsaTransactions {
   @CanIgnoreReturnValue
   public static <T> T bsaTransact(Callable<T> work) {
     verify(!isInTransaction(), "May only be used for top-level transactions.");
-    return tm().transact(work, TRANSACTION_REPEATABLE_READ);
+    return tm().transact(TRANSACTION_REPEATABLE_READ, work);
   }
 
   public static void bsaTransact(ThrowingRunnable work) {
     verify(!isInTransaction(), "May only be used for top-level transactions.");
-    tm().transact(work, TRANSACTION_REPEATABLE_READ);
+    tm().transact(TRANSACTION_REPEATABLE_READ, work);
   }
 
   @CanIgnoreReturnValue

core/src/main/java/google/registry/bsa/persistence/LabelDiffUpdates.java

@@ -67,6 +67,7 @@ public final class LabelDiffUpdates {
             labels.stream().collect(groupingBy(BlockLabel::labelType, toImmutableList())));
 
     tm().transact(
+            TRANSACTION_REPEATABLE_READ,
             () -> {
               for (Map.Entry<LabelType, ImmutableList<BlockLabel>> entry :
                   labelsByType.entrySet()) {
@@ -128,8 +129,7 @@ public final class LabelDiffUpdates {
                     break;
                 }
               }
-            },
-            TRANSACTION_REPEATABLE_READ);
+            });
     logger.atInfo().log("Processed %s of labels.", labels.size());
     return nonBlockedDomains.build();
   }

core/src/main/java/google/registry/bsa/persistence/RefreshScheduler.java

@@ -52,7 +52,7 @@ public class RefreshScheduler {
               }
               // No previously completed refreshes. Need start time of a completed download as
               // lower bound of refresh checks.
-              if (!mostRecentDownload.isPresent()) {
+              if (mostRecentDownload.isEmpty()) {
                 return Optional.empty();
               }
 

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

@@ -128,7 +128,7 @@ public class CloudDnsWriter extends BaseDnsWriter {
     // Return early if no DNS records should be published.
     // desiredRecordsBuilder is populated with an empty set to indicate that all existing records
     // should be deleted.
-    if (!domain.isPresent() || !domain.get().shouldPublishToDns()) {
+    if (domain.isEmpty() || !domain.get().shouldPublishToDns()) {
       desiredRecords.put(absoluteDomainName, ImmutableSet.of());
       return;
     }
@@ -192,7 +192,7 @@ public class CloudDnsWriter extends BaseDnsWriter {
     Optional<Host> host = loadByForeignKey(Host.class, hostName, clock.nowUtc());
 
     // Return early if the host is deleted.
-    if (!host.isPresent()) {
+    if (host.isEmpty()) {
       desiredRecords.put(absoluteHostName, ImmutableSet.of());
       return;
     }
@@ -247,7 +247,7 @@ public class CloudDnsWriter extends BaseDnsWriter {
     Optional<InternetDomainName> tld = Tlds.findTldForName(host);
 
     // Host not managed by our registry, no need to update DNS.
-    if (!tld.isPresent()) {
+    if (tld.isEmpty()) {
       logger.atSevere().log("publishHost called for invalid host '%s'.", hostName);
       return;
     }

core/src/main/java/google/registry/dns/writer/dnsupdate/DnsUpdateWriter.java

@@ -157,7 +157,7 @@ public class DnsUpdateWriter extends BaseDnsWriter {
     Optional<InternetDomainName> tld = Tlds.findTldForName(host);
 
     // host not managed by our registry, no need to update DNS.
-    if (!tld.isPresent()) {
+    if (tld.isEmpty()) {
       return;
     }
 

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

@@ -115,7 +115,7 @@ public class ExportPremiumTermsAction implements Runnable {
           "Skipping premium terms export for TLD %s because Drive folder isn't specified.", tldStr);
       return Optional.of("Skipping export because no Drive folder is associated with this TLD");
     }
-    if (!tld.getPremiumListName().isPresent()) {
+    if (tld.getPremiumListName().isEmpty()) {
       logger.atInfo().log("No premium terms to export for TLD '%s'.", tldStr);
       return Optional.of("No premium lists configured");
     }

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

@@ -63,7 +63,7 @@ class SyncRegistrarsSheet {
   boolean wereRegistrarsModified() {
     Optional<Cursor> cursor =
         tm().transact(() -> tm().loadByKeyIfPresent(Cursor.createGlobalVKey(SYNC_REGISTRAR_SHEET)));
-    DateTime lastUpdateTime = !cursor.isPresent() ? START_OF_TIME : cursor.get().getCursorTime();
+    DateTime lastUpdateTime = cursor.isEmpty() ? START_OF_TIME : cursor.get().getCursorTime();
     for (Registrar registrar : Registrar.loadAllCached()) {
       if (DateTimeUtils.isAtOrAfter(registrar.getLastUpdateTime(), lastUpdateTime)) {
         return true;

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

@@ -112,11 +112,11 @@ public class SyncRegistrarsSheetAction implements Runnable {
   @Override
   public void run() {
     final Optional<String> sheetId = Optional.ofNullable(idParam.orElse(idConfig.orElse(null)));
-    if (!sheetId.isPresent()) {
+    if (sheetId.isEmpty()) {
       Result.MISSINGNO.send(response, null);
       return;
     }
-    if (!idParam.isPresent()) {
+    if (idParam.isEmpty()) {
       if (!syncRegistrarsSheet.wereRegistrarsModified()) {
         Result.NOTMODIFIED.send(response, null);
         return;

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

@@ -81,6 +81,7 @@ public class FlowRunner {
       // TODO(mcilwain/weiminyu): Use transactReadOnly() here for TransactionalFlow and transact()
       //                          for MutatingFlow.
       return tm().transact(
+              isolationLevelOverride.orElse(null),
               () -> {
                 try {
                   EppOutput output = EppOutput.create(flowProvider.get().run());
@@ -96,8 +97,7 @@ public class FlowRunner {
                 } catch (EppException e) {
                   throw new EppRuntimeException(e);
                 }
-              },
-              isolationLevelOverride.orElse(null));
+              });
     } catch (DryRunException e) {
       return e.output;
     } catch (EppRuntimeException e) {

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

@@ -122,7 +122,7 @@ public final class ResourceFlowUtils {
   /** Check that the given AuthInfo is present for a resource being transferred. */
   public static void verifyAuthInfoPresentForResourceTransfer(Optional<AuthInfo> authInfo)
       throws EppException {
-    if (!authInfo.isPresent()) {
+    if (authInfo.isEmpty()) {
       throw new MissingTransferRequestAuthInfoException();
     }
   }
@@ -160,7 +160,7 @@ public final class ResourceFlowUtils {
         domain.getReferencedContacts().stream()
             .filter(key -> key.getKey().equals(authRepoId))
             .findFirst();
-    if (!foundContact.isPresent()) {
+    if (foundContact.isEmpty()) {
       throw new BadAuthInfoForResourceException();
     }
     // Check the authInfo against the contact.

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

@@ -105,7 +105,7 @@ public class TlsCredentials implements TransportCredentials {
     }
     // In the rare unexpected case that the client inet address wasn't passed along at all, then
     // by default deny access.
-    if (!clientInetAddr.isPresent()) {
+    if (clientInetAddr.isEmpty()) {
       logger.atWarning().log(
           "Authentication error: Missing IP address for registrar %s.", registrar.getRegistrarId());
       throw new BadRegistrarIpAddressException(clientInetAddr);
@@ -129,8 +129,8 @@ public class TlsCredentials implements TransportCredentials {
 
   @VisibleForTesting
   void validateCertificateHash(Registrar registrar) throws AuthenticationErrorException {
-    if (!registrar.getClientCertificateHash().isPresent()
-        && !registrar.getFailoverClientCertificateHash().isPresent()) {
+    if (registrar.getClientCertificateHash().isEmpty()
+        && registrar.getFailoverClientCertificateHash().isEmpty()) {
       if (requireSslCertificates) {
         throw new RegistrarCertificateNotConfiguredException();
       } else {
@@ -140,7 +140,7 @@ public class TlsCredentials implements TransportCredentials {
       }
     }
     // Check that the request included the certificate hash
-    if (!clientCertificateHash.isPresent()) {
+    if (clientCertificateHash.isEmpty()) {
       logger.atInfo().log(
           "Request from registrar %s did not include X-SSL-Certificate.",
           registrar.getRegistrarId());

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

@@ -75,7 +75,7 @@ public final class ContactTransferQueryFlow implements TransactionalFlow {
     }
     // Note that the authorization info on the command (if present) has already been verified. If
     // it's present, then the other checks are unnecessary.
-    if (!authInfo.isPresent()
+    if (authInfo.isEmpty()
         && !registrarId.equals(contact.getTransferData().getGainingRegistrarId())
         && !registrarId.equals(contact.getTransferData().getLosingRegistrarId())) {
       throw new NotAuthorizedToViewTransferException();

core/src/main/java/google/registry/flows/custom/DomainRenewFlowCustomLogic.java

@@ -14,6 +14,7 @@
 
 package google.registry.flows.custom;
 
+import com.google.auto.value.AutoBuilder;
 import com.google.auto.value.AutoValue;
 import com.google.common.collect.ImmutableList;
 import google.registry.flows.EppException;
@@ -81,30 +82,23 @@ public class DomainRenewFlowCustomLogic extends BaseFlowCustomLogic {
   }
 
   /** A class to encapsulate parameters for a call to {@link #afterValidation}. */
-  @AutoValue
-  public abstract static class AfterValidationParameters extends ImmutableObject {
-
-    public abstract Domain existingDomain();
-
-    public abstract int years();
-
-    public abstract DateTime now();
+  public record AfterValidationParameters(Domain existingDomain, int years, DateTime now) {
 
     public static Builder newBuilder() {
-      return new AutoValue_DomainRenewFlowCustomLogic_AfterValidationParameters.Builder();
+      return new AutoBuilder_DomainRenewFlowCustomLogic_AfterValidationParameters_Builder();
     }
 
     /** Builder for {@link AfterValidationParameters}. */
-    @AutoValue.Builder
-    public abstract static class Builder {
+    @AutoBuilder
+    public interface Builder {
 
-      public abstract Builder setExistingDomain(Domain existingDomain);
+      Builder setExistingDomain(Domain existingDomain);
 
-      public abstract Builder setYears(int years);
+      Builder setYears(int years);
 
-      public abstract Builder setNow(DateTime now);
+      Builder setNow(DateTime now);
 
-      public abstract AfterValidationParameters build();
+      AfterValidationParameters build();
     }
   }
 

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

@@ -26,6 +26,7 @@ import static google.registry.flows.domain.DomainFlowUtils.checkHasBillingAccoun
 import static google.registry.flows.domain.DomainFlowUtils.getReservationTypes;
 import static google.registry.flows.domain.DomainFlowUtils.handleFeeRequest;
 import static google.registry.flows.domain.DomainFlowUtils.isAnchorTenant;
+import static google.registry.flows.domain.DomainFlowUtils.isRegisterBsaCreate;
 import static google.registry.flows.domain.DomainFlowUtils.isReserved;
 import static google.registry.flows.domain.DomainFlowUtils.isValidReservedCreate;
 import static google.registry.flows.domain.DomainFlowUtils.validateDomainName;
@@ -218,7 +219,7 @@ public final class DomainCheckFlow implements TransactionalFlow {
               domainCheckResults,
               tldStates,
               allocationToken);
-      boolean isAvailable = !message.isPresent();
+      boolean isAvailable = message.isEmpty();
       checksBuilder.add(DomainCheck.create(isAvailable, domainName, message.orElse(null)));
       if (isAvailable) {
         availableDomains.add(domainName);
@@ -269,13 +270,13 @@ public final class DomainCheckFlow implements TransactionalFlow {
     if (tokenResult.isPresent()) {
       return tokenResult;
     }
-    if (bsaBlockedDomains.contains(domainName)) {
-      // TODO(weiminyu): extract to a constant for here and CheckApiAction.
-      // Excerpt from BSA's custom message. Max len 32 chars by EPP XML schema.
-      return Optional.of("Blocked by a GlobalBlock service");
-    } else {
+    if (isRegisterBsaCreate(domainName, allocationToken)
+        || !bsaBlockedDomains.contains(domainName)) {
       return Optional.empty();
     }
+    // TODO(weiminyu): extract to a constant for here and CheckApiAction.
+    // Excerpt from BSA's custom message. Max len 32 chars by EPP XML schema.
+    return Optional.of("Blocked by a GlobalBlock service");
   }
 
   /** Handle the fee check extension. */
@@ -288,7 +289,7 @@ public final class DomainCheckFlow implements TransactionalFlow {
       throws EppException {
     Optional<FeeCheckCommandExtension> feeCheckOpt =
         eppInput.getSingleExtension(FeeCheckCommandExtension.class);
-    if (!feeCheckOpt.isPresent()) {
+    if (feeCheckOpt.isEmpty()) {
       return ImmutableList.of(); // No fee checks were requested.
     }
     FeeCheckCommandExtension<?, ?> feeCheck = feeCheckOpt.get();

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

@@ -276,7 +276,7 @@ public final class DomainCreateFlow implements MutatingFlow {
             now,
             eppInput.getSingleExtension(AllocationTokenExtension.class));
     boolean defaultTokenUsed = false;
-    if (!allocationToken.isPresent()) {
+    if (allocationToken.isEmpty()) {
       allocationToken =
           DomainFlowUtils.checkForDefaultToken(
               tld, command.getDomainName(), CommandName.CREATE, registrarId, now);
@@ -330,7 +330,7 @@ public final class DomainCreateFlow implements MutatingFlow {
               .verifySignedMarks(launchCreate.get().getSignedMarks(), domainLabel, now)
               .getId();
     }
-    verifyNotBlockedByBsa(domainLabel, tld, now);
+    verifyNotBlockedByBsa(domainName, tld, now, allocationToken);
     flowCustomLogic.afterValidation(
         DomainCreateFlowCustomLogic.AfterValidationParameters.newBuilder()
             .setDomainName(domainName)
@@ -421,8 +421,7 @@ public final class DomainCreateFlow implements MutatingFlow {
           createNameCollisionOneTimePollMessage(targetId, domainHistory, registrarId, now));
     }
     entitiesToSave.add(domain, domainHistory);
-    if (allocationToken.isPresent()
-        && TokenType.SINGLE_USE.equals(allocationToken.get().getTokenType())) {
+    if (allocationToken.isPresent() && allocationToken.get().getTokenType().isOneTimeUse()) {
       entitiesToSave.add(
           allocationTokenFlowUtils.redeemToken(
               allocationToken.get(), domainHistory.getHistoryEntryId()));
@@ -520,7 +519,7 @@ public final class DomainCreateFlow implements MutatingFlow {
     if (behavior.equals(RegistrationBehavior.BYPASS_TLD_STATE)
         || behavior.equals(RegistrationBehavior.ANCHOR_TENANT)) {
       // Non-trademarked names with the state check bypassed are always available
-      if (!claimsList.getClaimKey(domainLabel).isPresent()) {
+      if (claimsList.getClaimKey(domainLabel).isEmpty()) {
         return;
       }
       if (!currentState.equals(START_DATE_SUNRISE)) {

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

@@ -27,6 +27,7 @@ import static com.google.common.collect.Sets.intersection;
 import static com.google.common.collect.Sets.union;
 import static google.registry.bsa.persistence.BsaLabelUtils.isLabelBlocked;
 import static google.registry.model.domain.Domain.MAX_REGISTRATION_YEARS;
+import static google.registry.model.domain.token.AllocationToken.TokenType.REGISTER_BSA;
 import static google.registry.model.tld.Tld.TldState.GENERAL_AVAILABILITY;
 import static google.registry.model.tld.Tld.TldState.PREDELEGATION;
 import static google.registry.model.tld.Tld.TldState.QUIET_PERIOD;
@@ -214,7 +215,7 @@ public class DomainFlowUtils {
       throw new DomainNameExistsAsTldException();
     }
     Optional<InternetDomainName> tldParsed = findTldForName(domainName);
-    if (!tldParsed.isPresent()) {
+    if (tldParsed.isEmpty()) {
       throw new TldDoesNotExistException(domainName.parent().toString());
     }
     if (domainName.parts().size() != tldParsed.get().parts().size() + 1) {
@@ -255,7 +256,7 @@ public class DomainFlowUtils {
     Optional<String> idnTableName =
         IDN_LABEL_VALIDATOR.findValidIdnTableForTld(
             domainName.parts().get(0), domainName.parent().toString());
-    if (!idnTableName.isPresent()) {
+    if (idnTableName.isEmpty()) {
       throw new InvalidIdnDomainLabelException();
     }
     return idnTableName.get();
@@ -265,9 +266,14 @@ public class DomainFlowUtils {
    * Verifies that the {@code domainLabel} is not blocked by any BSA block label for the given
    * {@code tld} at the specified time.
    */
-  public static void verifyNotBlockedByBsa(String domainLabel, Tld tld, DateTime now)
+  public static void verifyNotBlockedByBsa(
+      InternetDomainName domainName,
+      Tld tld,
+      DateTime now,
+      Optional<AllocationToken> allocationToken)
       throws DomainLabelBlockedByBsaException {
-    if (isBlockedByBsa(domainLabel, tld, now)) {
+    if (!isRegisterBsaCreate(domainName, allocationToken)
+        && isBlockedByBsa(domainName.parts().get(0), tld, now)) {
       throw new DomainLabelBlockedByBsaException();
     }
   }
@@ -311,6 +317,15 @@ public class DomainFlowUtils {
         && token.get().getDomainName().get().equals(domainName.toString());
   }
 
+  /** Returns whether a given domain create request may bypass the BSA block check. */
+  public static boolean isRegisterBsaCreate(
+      InternetDomainName domainName, Optional<AllocationToken> token) {
+    return token.isPresent()
+        && token.get().getTokenType().equals(REGISTER_BSA)
+        && token.get().getDomainName().isPresent()
+        && token.get().getDomainName().get().equals(domainName.toString());
+  }
+
   /** Check if the registrar running the flow has access to the TLD in question. */
   public static void checkAllowedAccessToTld(String registrarId, String tld) throws EppException {
     if (!Registrar.loadByRegistrarIdCached(registrarId).get().getAllowedTlds().contains(tld)) {
@@ -353,7 +368,7 @@ public class DomainFlowUtils {
       }
       ImmutableList<DomainDsData> invalidDigestTypes =
           dsData.stream()
-              .filter(ds -> !DigestType.fromWireValue(ds.getDigestType()).isPresent())
+              .filter(ds -> DigestType.fromWireValue(ds.getDigestType()).isEmpty())
               .collect(toImmutableList());
       if (!invalidDigestTypes.isEmpty()) {
         throw new InvalidDsRecordException(
@@ -801,7 +816,7 @@ public class DomainFlowUtils {
       FeesAndCredits feesAndCredits,
       boolean defaultTokenUsed)
       throws EppException {
-    if (feesAndCredits.hasAnyPremiumFees() && !feeCommand.isPresent()) {
+    if (feesAndCredits.hasAnyPremiumFees() && feeCommand.isEmpty()) {
       throw new FeesRequiredForPremiumNameException();
     }
     validateFeesAckedIfPresent(feeCommand, feesAndCredits, defaultTokenUsed);
@@ -822,7 +837,7 @@ public class DomainFlowUtils {
     // Check for the case where a fee command extension was required but not provided.
     // This only happens when the total fees are non-zero and include custom fees requiring the
     // extension.
-    if (!feeCommand.isPresent()) {
+    if (feeCommand.isEmpty()) {
       if (!feesAndCredits.getEapCost().isZero()) {
         throw new FeesRequiredDuringEarlyAccessProgramException(feesAndCredits.getEapCost());
       }
@@ -1034,7 +1049,7 @@ public class DomainFlowUtils {
   /** Validate the secDNS extension, if present. */
   static Optional<SecDnsCreateExtension> validateSecDnsExtension(
       Optional<SecDnsCreateExtension> secDnsCreate) throws EppException {
-    if (!secDnsCreate.isPresent()) {
+    if (secDnsCreate.isEmpty()) {
       return Optional.empty();
     }
     if (secDnsCreate.get().getDsData() == null) {

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

@@ -73,7 +73,6 @@ import google.registry.model.domain.fee.FeeTransformResponseExtension;
 import google.registry.model.domain.metadata.MetadataExtension;
 import google.registry.model.domain.rgp.GracePeriodStatus;
 import google.registry.model.domain.token.AllocationToken;
-import google.registry.model.domain.token.AllocationToken.TokenType;
 import google.registry.model.domain.token.AllocationTokenExtension;
 import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppcommon.StatusValue;
@@ -183,7 +182,7 @@ public final class DomainRenewFlow implements MutatingFlow {
             CommandName.RENEW,
             eppInput.getSingleExtension(AllocationTokenExtension.class));
     boolean defaultTokenUsed = false;
-    if (!allocationToken.isPresent()) {
+    if (allocationToken.isEmpty()) {
       allocationToken =
           DomainFlowUtils.checkForDefaultToken(
               tld, existingDomain.getDomainName(), CommandName.RENEW, registrarId, now);
@@ -258,8 +257,7 @@ public final class DomainRenewFlow implements MutatingFlow {
     ImmutableSet.Builder<ImmutableObject> entitiesToSave = new ImmutableSet.Builder<>();
     entitiesToSave.add(
         newDomain, domainHistory, explicitRenewEvent, newAutorenewEvent, newAutorenewPollMessage);
-    if (allocationToken.isPresent()
-        && TokenType.SINGLE_USE.equals(allocationToken.get().getTokenType())) {
+    if (allocationToken.isPresent() && allocationToken.get().getTokenType().isOneTimeUse()) {
       entitiesToSave.add(
           allocationTokenFlowUtils.redeemToken(
               allocationToken.get(), domainHistory.getHistoryEntryId()));

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

@@ -81,7 +81,7 @@ public final class DomainTransferQueryFlow implements TransactionalFlow {
     }
     // Note that the authorization info on the command (if present) has already been verified. If
     // it's present, then the other checks are unnecessary.
-    if (!authInfo.isPresent()
+    if (authInfo.isEmpty()
         && !registrarId.equals(transferData.getGainingRegistrarId())
         && !registrarId.equals(transferData.getLosingRegistrarId())) {
       throw new NotAuthorizedToViewTransferException();

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

@@ -201,7 +201,7 @@ public final class DomainTransferRequestFlow implements MutatingFlow {
     Optional<FeesAndCredits> feesAndCredits;
     if (period.getValue() == 0) {
       feesAndCredits = Optional.empty();
-    } else if (!existingDomain.getCurrentBulkToken().isPresent()) {
+    } else if (existingDomain.getCurrentBulkToken().isEmpty()) {
       feesAndCredits =
           Optional.of(pricingLogic.getTransferPrice(tld, targetId, now, existingBillingRecurrence));
     } else {

core/src/main/java/google/registry/flows/domain/token/AllocationTokenFlowUtils.java

@@ -36,7 +36,6 @@ import google.registry.model.domain.fee.FeeQueryCommandExtensionItem.CommandName
 import google.registry.model.domain.token.AllocationToken;
 import google.registry.model.domain.token.AllocationToken.TokenBehavior;
 import google.registry.model.domain.token.AllocationToken.TokenStatus;
-import google.registry.model.domain.token.AllocationToken.TokenType;
 import google.registry.model.domain.token.AllocationTokenExtension;
 import google.registry.model.reporting.HistoryEntry.HistoryEntryId;
 import google.registry.model.tld.Tld;
@@ -105,8 +104,7 @@ public class AllocationTokenFlowUtils {
   /** Redeems a SINGLE_USE {@link AllocationToken}, returning the redeemed copy. */
   public AllocationToken redeemToken(AllocationToken token, HistoryEntryId redemptionHistoryId) {
     checkArgument(
-        TokenType.SINGLE_USE.equals(token.getTokenType()),
-        "Only SINGLE_USE tokens can be marked as redeemed");
+        token.getTokenType().isOneTimeUse(), "Only SINGLE_USE tokens can be marked as redeemed");
     return token.asBuilder().setRedemptionHistoryId(redemptionHistoryId).build();
   }
 
@@ -184,7 +182,7 @@ public class AllocationTokenFlowUtils {
     maybeTokenEntity =
         tm().transact(() -> tm().loadByKeyIfPresent(VKey.create(AllocationToken.class, token)));
 
-    if (!maybeTokenEntity.isPresent()) {
+    if (maybeTokenEntity.isEmpty()) {
       throw new InvalidAllocationTokenException();
     }
     if (maybeTokenEntity.get().isRedeemed()) {
@@ -201,7 +199,7 @@ public class AllocationTokenFlowUtils {
       DateTime now,
       Optional<AllocationTokenExtension> extension)
       throws EppException {
-    if (!extension.isPresent()) {
+    if (extension.isEmpty()) {
       return Optional.empty();
     }
     AllocationToken tokenEntity = loadToken(extension.get().getAllocationToken());
@@ -224,7 +222,7 @@ public class AllocationTokenFlowUtils {
       CommandName commandName,
       Optional<AllocationTokenExtension> extension)
       throws EppException {
-    if (!extension.isPresent()) {
+    if (extension.isEmpty()) {
       return Optional.empty();
     }
     AllocationToken tokenEntity = loadToken(extension.get().getAllocationToken());
@@ -256,7 +254,7 @@ public class AllocationTokenFlowUtils {
 
   public static Domain maybeApplyBulkPricingRemovalToken(
       Domain domain, Optional<AllocationToken> allocationToken) {
-    if (!allocationToken.isPresent()
+    if (allocationToken.isEmpty()
         || !TokenBehavior.REMOVE_BULK_PRICING.equals(allocationToken.get().getTokenBehavior())) {
       return domain;
     }

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

@@ -81,7 +81,7 @@ public class HostFlowUtils {
   public static Optional<Domain> lookupSuperordinateDomain(
       InternetDomainName hostName, DateTime now) throws EppException {
     Optional<InternetDomainName> tld = findTldForName(hostName);
-    if (!tld.isPresent()) {
+    if (tld.isEmpty()) {
       // This is an host on a TLD we don't run, therefore obviously external, so we are done.
       return Optional.empty();
     }
@@ -91,7 +91,7 @@ public class HostFlowUtils {
             .skip(hostName.parts().size() - (tld.get().parts().size() + 1))
             .collect(joining("."));
     Optional<Domain> superordinateDomain = loadByForeignKey(Domain.class, domainName, now);
-    if (!superordinateDomain.isPresent() || !isActive(superordinateDomain.get(), now)) {
+    if (superordinateDomain.isEmpty() || !isActive(superordinateDomain.get(), now)) {
       throw new SuperordinateDomainDoesNotExistException(domainName);
     }
     return superordinateDomain;

core/src/main/java/google/registry/flows/picker/FlowPicker.java

@@ -148,23 +148,25 @@ public class FlowPicker {
    * <p>This provider must be tried before {@link #RESOURCE_CRUD_FLOW_PROVIDER}. Otherwise, the
    * regular domain update flow will match first.
    */
-  private static final FlowProvider DOMAIN_RESTORE_FLOW_PROVIDER = new FlowProvider() {
-    @Override
-    Class<? extends Flow> get(
-        EppInput eppInput, InnerCommand innerCommand, ResourceCommand resourceCommand) {
-      if (!(resourceCommand instanceof DomainCommand.Update)) {
-        return null;
-      }
-      Optional<RgpUpdateExtension> rgpUpdateExtension =
-          eppInput.getSingleExtension(RgpUpdateExtension.class);
-      if (!rgpUpdateExtension.isPresent()) {
-        return null;
-      }
-      // Restore command with an op of "report" is not currently supported.
-      return (rgpUpdateExtension.get().getRestoreCommand().getRestoreOp() == RestoreOp.REQUEST)
-          ? DomainRestoreRequestFlow.class
-          : UnimplementedRestoreFlow.class;
-    }};
+  private static final FlowProvider DOMAIN_RESTORE_FLOW_PROVIDER =
+      new FlowProvider() {
+        @Override
+        Class<? extends Flow> get(
+            EppInput eppInput, InnerCommand innerCommand, ResourceCommand resourceCommand) {
+          if (!(resourceCommand instanceof DomainCommand.Update)) {
+            return null;
+          }
+          Optional<RgpUpdateExtension> rgpUpdateExtension =
+              eppInput.getSingleExtension(RgpUpdateExtension.class);
+          if (rgpUpdateExtension.isEmpty()) {
+            return null;
+          }
+          // Restore command with an op of "report" is not currently supported.
+          return (rgpUpdateExtension.get().getRestoreCommand().getRestoreOp() == RestoreOp.REQUEST)
+              ? DomainRestoreRequestFlow.class
+              : UnimplementedRestoreFlow.class;
+        }
+      };
 
   /**
    * The claims check flow is keyed on the type of the {@link ResourceCommand} and on having the
@@ -180,7 +182,7 @@ public class FlowPicker {
           }
           Optional<LaunchCheckExtension> launchCheck =
               eppInput.getSingleExtension(LaunchCheckExtension.class);
-          if (!launchCheck.isPresent()
+          if (launchCheck.isEmpty()
               || CheckType.AVAILABILITY.equals(launchCheck.get().getCheckType())) {
             // We don't distinguish between registry phases for "avail", so don't bother checking
             // phase.

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

@@ -86,7 +86,7 @@ public final class PollAckFlow implements MutatingFlow {
     // it as if it doesn't exist yet. Same for if the message ID year isn't the same as the actual
     // poll message's event time (that means they're passing in an old already-acked ID).
     Optional<PollMessage> maybePollMessage = tm().loadByKeyIfPresent(pollMessageKey);
-    if (!maybePollMessage.isPresent()
+    if (maybePollMessage.isEmpty()
         || !isBeforeOrAt(maybePollMessage.get().getEventTime(), now)
         || !makePollMessageExternalId(maybePollMessage.get()).equals(messageId)) {
       throw new MessageDoesNotExistException(messageId);

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

@@ -66,7 +66,7 @@ public final class PollRequestFlow implements TransactionalFlow {
     // Return the oldest message from the queue.
     DateTime now = tm().getTransactionTime();
     Optional<PollMessage> maybePollMessage = getFirstPollMessage(registrarId, now);
-    if (!maybePollMessage.isPresent()) {
+    if (maybePollMessage.isEmpty()) {
       return responseBuilder.setResultFromCode(SUCCESS_WITH_NO_MESSAGES).build();
     }
     PollMessage pollMessage = maybePollMessage.get();

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

@@ -123,7 +123,7 @@ public class LoginFlow implements MutatingFlow {
       serviceExtensionUrisBuilder.add(uri);
     }
     Optional<Registrar> registrar = Registrar.loadByRegistrarIdCached(login.getClientId());
-    if (!registrar.isPresent()) {
+    if (registrar.isEmpty()) {
       throw new BadRegistrarIdException(login.getClientId());
     }
 
@@ -155,7 +155,7 @@ public class LoginFlow implements MutatingFlow {
                   });
       // Load fresh from database (bypassing the cache) to ensure we don't save stale data.
       Optional<Registrar> freshRegistrar = Registrar.loadByRegistrarId(login.getClientId());
-      if (!freshRegistrar.isPresent()) {
+      if (freshRegistrar.isEmpty()) {
         throw new BadRegistrarIdException(login.getClientId());
       }
       tm().put(freshRegistrar.get().asBuilder().setPassword(newPassword).build());

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

@@ -539,7 +539,7 @@ public class DomainBase extends EppResource
     for (GracePeriod gracePeriod : almostBuilt.getGracePeriods()) {
       if (isBeforeOrAt(gracePeriod.getExpirationTime(), now)) {
         builder.removeGracePeriod(gracePeriod);
-        if (!newLastEppUpdateTime.isPresent()
+        if (newLastEppUpdateTime.isEmpty()
             || isBeforeOrAt(newLastEppUpdateTime.get(), gracePeriod.getExpirationTime())) {
           newLastEppUpdateTime = Optional.of(gracePeriod.getExpirationTime());
         }

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

@@ -219,7 +219,7 @@ public final class RegistryLock extends UpdateAutoTimestampEntity implements Bui
 
   /** Returns true iff the lock was requested >= 1 hour ago and has not been verified. */
   public boolean isLockRequestExpired(DateTime now) {
-    return !getLockCompletionTime().isPresent()
+    return getLockCompletionTime().isEmpty()
         && isBeforeOrAt(getLockRequestTime(), now.minusHours(1));
   }
 
@@ -227,7 +227,7 @@ public final class RegistryLock extends UpdateAutoTimestampEntity implements Bui
   public boolean isUnlockRequestExpired(DateTime now) {
     Optional<DateTime> unlockRequestTimestamp = getUnlockRequestTime();
     return unlockRequestTimestamp.isPresent()
-        && !getUnlockCompletionTime().isPresent()
+        && getUnlockCompletionTime().isEmpty()
         && isBeforeOrAt(unlockRequestTimestamp.get(), now.minusHours(1));
   }
 

core/src/main/java/google/registry/model/domain/token/AllocationToken.java

@@ -22,6 +22,7 @@ import static google.registry.model.domain.token.AllocationToken.TokenStatus.CAN
 import static google.registry.model.domain.token.AllocationToken.TokenStatus.ENDED;
 import static google.registry.model.domain.token.AllocationToken.TokenStatus.NOT_STARTED;
 import static google.registry.model.domain.token.AllocationToken.TokenStatus.VALID;
+import static google.registry.model.domain.token.AllocationToken.TokenType.REGISTER_BSA;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.forceEmptyToNull;
 import static google.registry.util.CollectionUtils.nullToEmptyImmutableCopy;
@@ -120,18 +121,37 @@ public class AllocationToken extends UpdateAutoTimestampEntity implements Builda
   /** Type of the token that indicates how and where it should be used. */
   public enum TokenType {
     /** Token used for bulk pricing */
-    BULK_PRICING,
+    BULK_PRICING(/* isOneTimeUse= */ false),
     /** Token saved on a TLD to use if no other token is passed from the client */
-    DEFAULT_PROMO,
+    DEFAULT_PROMO(/* isOneTimeUse= */ false),
     /** This is the old name for what is now BULK_PRICING. */
     // TODO(sarahbot@): Remove this type once all tokens of this type have been scrubbed from the
     // database
     @Deprecated
-    PACKAGE,
+    PACKAGE(/* isOneTimeUse= */ false),
     /** Invalid after use */
-    SINGLE_USE,
+    SINGLE_USE(/* isOneTimeUse= */ true),
     /** Do not expire after use */
-    UNLIMITED_USE,
+    UNLIMITED_USE(/* isOneTimeUse= */ false),
+    /**
+     * Allows bypassing the BSA check during domain creation, otherwise has the same semantics as
+     * {@link #SINGLE_USE}.
+     *
+     * <p>This token applies to a single domain only. If the domain is not blocked by BSA at the
+     * redemption time this token is processed like {@code SINGLE_USE}, as mentioned above.
+     */
+    REGISTER_BSA(/* isOneTimeUse= */ true);
+
+    private final boolean isOneTimeUse;
+
+    private TokenType(boolean isOneTimeUse) {
+      this.isOneTimeUse = isOneTimeUse;
+    }
+
+    /** Returns true if token should be invalidated after use. */
+    public boolean isOneTimeUse() {
+      return this.isOneTimeUse;
+    }
   }
 
   /**
@@ -361,12 +381,11 @@ public class AllocationToken extends UpdateAutoTimestampEntity implements Builda
               || !getInstance().discountPremiums,
           "Bulk tokens cannot discount premium names");
       checkArgument(
-          getInstance().domainName == null || TokenType.SINGLE_USE.equals(getInstance().tokenType),
-          "Domain name can only be specified for SINGLE_USE tokens");
+          getInstance().domainName == null || getInstance().tokenType.isOneTimeUse(),
+          "Domain name can only be specified for SINGLE_USE or REGISTER_BSA tokens");
       checkArgument(
-          getInstance().redemptionHistoryId == null
-              || TokenType.SINGLE_USE.equals(getInstance().tokenType),
-          "Redemption history entry can only be specified for SINGLE_USE tokens");
+          getInstance().redemptionHistoryId == null || getInstance().tokenType.isOneTimeUse(),
+          "Redemption history entry can only be specified for SINGLE_USE or REGISTER_BSA tokens");
       checkArgument(
           getInstance().tokenType != TokenType.BULK_PRICING
               || (getInstance().allowedClientIds != null
@@ -378,6 +397,10 @@ public class AllocationToken extends UpdateAutoTimestampEntity implements Builda
       checkArgument(
           getInstance().discountFraction > 0 || getInstance().discountYears == 1,
           "Discount years can only be specified along with a discount fraction");
+      if (getInstance().getTokenType().equals(REGISTER_BSA)) {
+        checkArgumentNotNull(
+            getInstance().domainName, "REGISTER_BSA tokens must be tied to a domain");
+      }
       if (getInstance().registrationBehavior.equals(RegistrationBehavior.ANCHOR_TENANT)) {
         checkArgumentNotNull(
             getInstance().domainName, "ANCHOR_TENANT tokens must be tied to a domain");

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

@@ -19,7 +19,6 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 import static google.registry.util.DateTimeUtils.isAtOrAfter;
 import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
 
-import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Strings;
 import com.google.common.flogger.FluentLogger;
@@ -129,26 +128,11 @@ public class Lock extends ImmutableObject implements Serializable {
     return String.format("%s-%s", scope, resourceName);
   }
 
-  @AutoValue
-  abstract static class AcquireResult {
-    public abstract DateTime transactionTime();
-
-    @Nullable
-    public abstract Lock existingLock();
-
-    @Nullable
-    public abstract Lock newLock();
-
-    public abstract LockState lockState();
-
-    public static AcquireResult create(
-        DateTime transactionTime,
-        @Nullable Lock existingLock,
-        @Nullable Lock newLock,
-        LockState lockState) {
-      return new AutoValue_Lock_AcquireResult(transactionTime, existingLock, newLock, lockState);
-    }
-  }
+  record AcquireResult(
+      DateTime transactionTime,
+      @Nullable Lock existingLock,
+      @Nullable Lock newLock,
+      LockState lockState) {}
 
   private static void logAcquireResult(AcquireResult acquireResult) {
     try {
@@ -204,13 +188,13 @@ public class Lock extends ImmutableObject implements Serializable {
             lockState = LockState.TIMED_OUT;
           } else {
             lockState = LockState.IN_USE;
-            return AcquireResult.create(now, lock, null, lockState);
+            return new AcquireResult(now, lock, null, lockState);
           }
 
           Lock newLock = create(resourceName, scope, now, leaseLength);
           tm().put(newLock);
 
-          return AcquireResult.create(now, lock, newLock, lockState);
+          return new AcquireResult(now, lock, newLock, lockState);
         };
     AcquireResult acquireResult = tm().transact(lockAcquirer);
 

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

@@ -46,7 +46,7 @@ public class ServerSecret extends CrossTldSingleton {
               // Make sure we're in a transaction and attempt to load any existing secret, then
               // create it if it's absent.
               Optional<ServerSecret> secret = tm().loadSingleton(ServerSecret.class);
-              if (!secret.isPresent()) {
+              if (secret.isEmpty()) {
                 secret = Optional.of(create(UUID.randomUUID()));
                 tm().insert(secret.get());
               }

core/src/main/java/google/registry/model/tld/Tld.java

@@ -74,6 +74,7 @@ import google.registry.tldconfig.idn.IdnTableEnum;
 import google.registry.util.Idn;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Predicate;
@@ -458,6 +459,8 @@ public class Tld extends ImmutableObject implements Buildable, UnsafeSerializabl
   @JsonDeserialize(using = CurrencyDeserializer.class)
   CurrencyUnit currency = DEFAULT_CURRENCY;
 
+  // TODO(sarahbot@): Remove this field and make createBillingCostTransitions not-null once all TLDs
+  // are populated with a create cost transition map
   /** The per-year billing cost for registering a new domain name. */
   @Type(type = JodaMoneyType.TYPE_NAME)
   @Columns(
@@ -467,6 +470,12 @@ public class Tld extends ImmutableObject implements Buildable, UnsafeSerializabl
       })
   Money createBillingCost = DEFAULT_CREATE_BILLING_COST;
 
+  // TODO(sarahbot@): Make this field not null and add a default value once field is populated on
+  // all existing TLDs
+  /** A property that transitions to different create billing costs at different times. */
+  @JsonDeserialize(using = TimedTransitionPropertyMoneyDeserializer.class)
+  TimedTransitionProperty<Money> createBillingCostTransitions;
+
   /** The one-time billing cost for restoring a domain name from the redemption grace period. */
   @Type(type = JodaMoneyType.TYPE_NAME)
   @Columns(
@@ -676,6 +685,13 @@ public class Tld extends ImmutableObject implements Buildable, UnsafeSerializabl
     return createBillingCost;
   }
 
+  public ImmutableSortedMap<DateTime, Money> getCreateBillingCostTransitions() {
+    return Objects.requireNonNullElseGet(
+            createBillingCostTransitions,
+            () -> TimedTransitionProperty.withInitialValue(getCreateBillingCost()))
+        .toValueMap();
+  }
+
   /**
    * Returns the add-on cost of a domain restore (the flat tld-wide fee charged in addition to one
    * year of renewal for that name).
@@ -959,6 +975,17 @@ public class Tld extends ImmutableObject implements Buildable, UnsafeSerializabl
       return this;
     }
 
+    public Builder setCreateBillingCostTransitions(
+        ImmutableSortedMap<DateTime, Money> createCostsMap) {
+      checkArgumentNotNull(createCostsMap, "Create billing costs map cannot be null");
+      checkArgument(
+          createCostsMap.values().stream().allMatch(Money::isPositiveOrZero),
+          "Create billing cost cannot be negative");
+      getInstance().createBillingCostTransitions =
+          TimedTransitionProperty.fromValueMap(createCostsMap);
+      return this;
+    }
+
     public Builder setReservedListsByName(Set<String> reservedListNames) {
       // TODO(b/309175133): forbid if enrolled with BSA
       checkArgument(reservedListNames != null, "reservedListNames must not be null");
@@ -1131,6 +1158,10 @@ public class Tld extends ImmutableObject implements Buildable, UnsafeSerializabl
       // here to catch cases where we loaded an invalid TimedTransitionProperty from the database
       // and cloned it into a new builder, to block re-building a Tld in an invalid state.
       instance.tldStateTransitions.checkValidity();
+      // TODO(sarahbot@): Remove null check when createBillingCostTransitions field is made not-null
+      if (instance.createBillingCostTransitions != null) {
+        instance.createBillingCostTransitions.checkValidity();
+      }
       instance.renewBillingCostTransitions.checkValidity();
       instance.eapFeeSchedule.checkValidity();
       // All costs must be in the expected currency.

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

@@ -115,7 +115,7 @@ public final class PremiumListDao {
    */
   public static Optional<Money> getPremiumPrice(String premiumListName, String label) {
     Optional<PremiumList> maybeLoadedList = getLatestRevision(premiumListName);
-    if (!maybeLoadedList.isPresent()) {
+    if (maybeLoadedList.isEmpty()) {
       return Optional.empty();
     }
     PremiumList loadedList = maybeLoadedList.get();

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

@@ -14,7 +14,6 @@
 
 package google.registry.module;
 
-import com.google.appengine.api.LifecycleManager;
 import com.google.common.flogger.FluentLogger;
 import com.google.monitoring.metrics.MetricReporter;
 import dagger.Lazy;
@@ -46,25 +45,26 @@ public class ServletBase extends HttpServlet {
   public void init() {
     Security.addProvider(new BouncyCastleProvider());
 
-    // If metric reporter failed to instantiate for any reason (bad keyring, bad json credential,
-    // etc), we log the error but keep the main thread running. Also the shutdown hook will only be
-    // registered if metric reporter starts up correctly.
+    // If the metric reporter failed to instantiate for any reason (bad keyring, bad json
+    // credential, etc.), we log the error but keep the main thread running. Also, the shutdown hook
+    // will only be registered if the metric reporter starts up correctly.
     try {
       metricReporter.get().startAsync().awaitRunning(java.time.Duration.ofSeconds(10));
       logger.atInfo().log("Started up MetricReporter.");
-      LifecycleManager.getInstance()
-          .setShutdownHook(
-              () -> {
-                try {
-                  metricReporter
-                      .get()
-                      .stopAsync()
-                      .awaitTerminated(java.time.Duration.ofSeconds(10));
-                  logger.atInfo().log("Shut down MetricReporter.");
-                } catch (TimeoutException e) {
-                  logger.atSevere().withCause(e).log("Failed to stop MetricReporter.");
-                }
-              });
+      Runtime.getRuntime()
+          .addShutdownHook(
+              new Thread(
+                  () -> {
+                    try {
+                      metricReporter
+                          .get()
+                          .stopAsync()
+                          .awaitTerminated(java.time.Duration.ofSeconds(10));
+                      logger.atInfo().log("Shut down MetricReporter.");
+                    } catch (TimeoutException e) {
+                      logger.atSevere().withCause(e).log("Failed to stop MetricReporter.");
+                    }
+                  }));
     } catch (Exception e) {
       logger.atSevere().withCause(e).log("Failed to initialize MetricReporter.");
     }

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

@@ -175,14 +175,14 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   public <T> T reTransact(Callable<T> work) {
     // This prevents inner transaction from retrying, thus avoiding a cascade retry effect.
     if (inTransaction()) {
-      return transactNoRetry(work, null);
+      return transactNoRetry(null, work);
     }
     return retrier.callWithRetry(
-        () -> transactNoRetry(work, null), JpaRetries::isFailedTxnRetriable);
+        () -> transactNoRetry(null, work), JpaRetries::isFailedTxnRetriable);
   }
 
   @Override
-  public <T> T transact(Callable<T> work, TransactionIsolationLevel isolationLevel) {
+  public <T> T transact(TransactionIsolationLevel isolationLevel, Callable<T> work) {
     if (inTransaction()) {
       if (!getHibernateAllowNestedTransactions()) {
         throw new IllegalStateException(NESTED_TRANSACTION_MESSAGE);
@@ -192,20 +192,25 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
         logger.atWarning().withStackTrace(StackSize.MEDIUM).log(NESTED_TRANSACTION_MESSAGE);
       }
       // This prevents inner transaction from retrying, thus avoiding a cascade retry effect.
-      return transactNoRetry(work, isolationLevel);
+      return transactNoRetry(isolationLevel, work);
     }
     return retrier.callWithRetry(
-        () -> transactNoRetry(work, isolationLevel), JpaRetries::isFailedTxnRetriable);
+        () -> transactNoRetry(isolationLevel, work), JpaRetries::isFailedTxnRetriable);
   }
 
   @Override
   public <T> T transact(Callable<T> work) {
-    return transact(work, null);
+    return transact(null, work);
+  }
+
+  @Override
+  public <T> T transactNoRetry(Callable<T> work) {
+    return transactNoRetry(null, work);
   }
 
   @Override
   public <T> T transactNoRetry(
-      Callable<T> work, @Nullable TransactionIsolationLevel isolationLevel) {
+      @Nullable TransactionIsolationLevel isolationLevel, Callable<T> work) {
     if (inTransaction()) {
       // This check will no longer be necessary when the transact() method always throws
       // inside a nested transaction, as the only way to pass a non-null isolation level
@@ -266,18 +271,18 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public void transact(ThrowingRunnable work, TransactionIsolationLevel isolationLevel) {
+  public void transact(TransactionIsolationLevel isolationLevel, ThrowingRunnable work) {
     transact(
+        isolationLevel,
         () -> {
           work.run();
           return null;
-        },
-        isolationLevel);
+        });
   }
 
   @Override
   public void transact(ThrowingRunnable work) {
-    transact(work, null);
+    transact(null, work);
   }
 
   @Override

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

@@ -61,7 +61,19 @@ public interface TransactionManager {
    * Executes the work in a transaction at the given {@link TransactionIsolationLevel} and returns
    * the result.
    */
-  <T> T transact(Callable<T> work, TransactionIsolationLevel isolationLevel);
+  <T> T transact(TransactionIsolationLevel isolationLevel, Callable<T> work);
+
+  /**
+   * Executes the work in a transaction and returns the result, without retrying upon retryable
+   * exceptions.
+   *
+   * <p>This method should only be used when the transaction contains side effects that are not
+   * rolled back by the transaction manager, for example in {@link
+   * google.registry.beam.common.RegistryJpaIO} where the results from a query are streamed to the
+   * next transformation inside a transaction, as the result stream has to materialize to a list
+   * outside a transaction and doing so would greatly affect the parallelism of the pipeline.
+   */
+  <T> T transactNoRetry(Callable<T> work);
 
   /**
    * Executes the work in a transaction at the given {@link TransactionIsolationLevel} and returns
@@ -73,7 +85,7 @@ public interface TransactionManager {
    * next transformation inside a transaction, as the result stream has to materialize to a list
    * outside a transaction and doing so would greatly affect the parallelism of the pipeline.
    */
-  <T> T transactNoRetry(Callable<T> work, TransactionIsolationLevel isolationLevel);
+  <T> T transactNoRetry(TransactionIsolationLevel isolationLevel, Callable<T> work);
 
   /**
    * Executes the work in a (potentially wrapped) transaction and returns the result.
@@ -95,7 +107,7 @@ public interface TransactionManager {
   void transact(ThrowingRunnable work);
 
   /** Executes the work in a transaction at the given {@link TransactionIsolationLevel}. */
-  void transact(ThrowingRunnable work, TransactionIsolationLevel isolationLevel);
+  void transact(TransactionIsolationLevel isolationLevel, ThrowingRunnable work);
 
   /**
    * Executes the work in a (potentially wrapped) transaction and returns the result.

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

@@ -387,7 +387,7 @@ abstract class AbstractJsonableObject implements Jsonable {
    */
   static void verifyAllowedJsonKeyName(String name, @Nullable Member member, Class<?> clazz) {
     Optional<ImmutableSet<String>> allowedFieldNames = getNameRestriction(clazz);
-    if (!allowedFieldNames.isPresent()) {
+    if (allowedFieldNames.isEmpty()) {
       return;
     }
     checkState(
@@ -416,7 +416,7 @@ abstract class AbstractJsonableObject implements Jsonable {
       // We ignore any Optional that are empty, as if they didn't exist at all
       if (object instanceof Optional) {
         Optional<?> optional = (Optional<?>) object;
-        if (!optional.isPresent()) {
+        if (optional.isEmpty()) {
           return;
         }
         object = optional.get();

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

@@ -64,7 +64,7 @@ public class RdapDomainAction extends RdapActionBase {
             Domain.class,
             pathSearchString,
             shouldIncludeDeleted() ? START_OF_TIME : rdapJsonFormatter.getRequestTime());
-    if (!domain.isPresent() || !isAuthorized(domain.get())) {
+    if (domain.isEmpty() || !isAuthorized(domain.get())) {
       // RFC7480 5.3 - if the server wishes to respond that it doesn't have data satisfying the
       // query, it MUST reply with 404 response code.
       //

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

@@ -342,7 +342,7 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
               Host.class,
               partialStringQuery.getInitialString(),
               shouldIncludeDeleted() ? START_OF_TIME : getRequestTime());
-      return (!host.isPresent()
+      return (host.isEmpty()
               || !desiredRegistrar.get().equals(host.get().getPersistedCurrentSponsorRegistrarId()))
           ? ImmutableList.of()
           : ImmutableList.of(host.get().createVKey());

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

@@ -121,7 +121,7 @@ public class RdapEntitySearchAction extends RdapSearchActionBase {
 
     // Check the subtype.
     Subtype subtype;
-    if (!subtypeParam.isPresent() || subtypeParam.get().equalsIgnoreCase("all")) {
+    if (subtypeParam.isEmpty() || subtypeParam.get().equalsIgnoreCase("all")) {
       subtype = Subtype.ALL;
     } else if (subtypeParam.get().equalsIgnoreCase("contacts")) {
       subtype = Subtype.CONTACTS;
@@ -133,7 +133,7 @@ public class RdapEntitySearchAction extends RdapSearchActionBase {
 
     CursorType cursorType;
     Optional<String> cursorQueryString;
-    if (!cursorString.isPresent()) {
+    if (cursorString.isEmpty()) {
       cursorType = CursorType.NONE;
       cursorQueryString = Optional.empty();
     } else {

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

@@ -101,7 +101,7 @@ public final class RdapModule {
   @Provides
   static RdapAuthorization provideRdapAuthorization(
       AuthResult authResult, AuthenticatedRegistrarAccessor registrarAccessor) {
-    if (!authResult.userAuthInfo().isPresent()) {
+    if (authResult.userAuthInfo().isEmpty()) {
       return RdapAuthorization.PUBLIC_AUTHORIZATION;
     }
     UserAuthInfo userAuthInfo = authResult.userAuthInfo().get();

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

@@ -66,7 +66,7 @@ public class RdapNameserverAction extends RdapActionBase {
             Host.class,
             pathSearchString,
             shouldIncludeDeleted() ? START_OF_TIME : getRequestTime());
-    if (!host.isPresent() || !isAuthorized(host.get())) {
+    if (host.isEmpty() || !isAuthorized(host.get())) {
       // RFC7480 5.3 - if the server wishes to respond that it doesn't have data satisfying the
       // query, it MUST reply with 404 response code.
       //

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

@@ -175,7 +175,7 @@ public class RdapNameserverSearchAction extends RdapSearchActionBase {
       RdapSearchPattern partialStringQuery) {
     Optional<Domain> domain =
         loadByForeignKeyCached(Domain.class, partialStringQuery.getSuffix(), getRequestTime());
-    if (!domain.isPresent()) {
+    if (domain.isEmpty()) {
       // Don't allow wildcards with suffixes which are not domains we manage. That would risk a
       // table scan in many easily foreseeable cases. The user might ask for ns*.zombo.com,
       // forcing us to query for all hosts beginning with ns, then filter for those ending in

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

@@ -121,7 +121,7 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
    */
   protected boolean shouldBeVisible(EppResource eppResource) {
     return isAuthorized(eppResource)
-        && (!registrarParam.isPresent()
+        && (registrarParam.isEmpty()
             || registrarParam.get().equals(eppResource.getPersistedCurrentSponsorRegistrarId()));
   }
 
@@ -135,7 +135,7 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
    */
   protected boolean shouldBeVisible(Registrar registrar) {
     return isAuthorized(registrar)
-        && (!registrarParam.isPresent() || registrarParam.get().equals(registrar.getRegistrarId()));
+        && (registrarParam.isEmpty() || registrarParam.get().equals(registrar.getRegistrarId()));
   }
 
   /**

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

@@ -99,7 +99,7 @@ public final class RdeReportAction implements Runnable, EscrowTask {
         RdeRevision.getCurrentRevision(tld, watermark, FULL)
             .orElseThrow(
                 () -> new IllegalStateException("RdeRevision was not set on generated deposit"));
-    if (!prefix.isPresent()) {
+    if (prefix.isEmpty()) {
       prefix = Optional.of(findMostRecentPrefixForWatermark(watermark, bucket, tld, gcsUtils));
     }
     String name = prefix.get() + RdeNamingUtils.makeRydeFilename(tld, watermark, FULL, 1, revision);

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

@@ -23,7 +23,6 @@ import static google.registry.util.DomainNameUtils.canonicalizeHostname;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import com.google.api.client.http.HttpMethods;
-import com.google.appengine.api.urlfetch.HTTPResponse;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
 import google.registry.config.RegistryConfig.Config;
@@ -107,7 +106,7 @@ public class RdeReporter {
   }
 
   /**
-   * Unmarshals IIRDEA XML result object from {@link HTTPResponse} payload.
+   * Unmarshall IIRDEA XML result object from HTTP response payload.
    *
    * @see <a
    *     href="http://tools.ietf.org/html/draft-lozano-icann-registry-interfaces-05#section-4.1">

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

@@ -352,7 +352,7 @@ public final class RdeStagingAction implements Runnable {
   }
 
   private ImmutableSetMultimap<String, PendingDeposit> getManualPendingDeposits() {
-    if (!directory.isPresent()) {
+    if (directory.isEmpty()) {
       throw new BadRequestException("Directory parameter required in manual operation");
     }
     if (directory.get().startsWith("/")) {

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

@@ -138,7 +138,7 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
   public void runWithLock(final DateTime watermark) throws Exception {
     // If a prefix is not provided,try to determine the prefix. This should only happen when the RDE
     // upload cron job runs to catch up any un-retried (i. e. expected) RDE failures.
-    if (!prefix.isPresent()) {
+    if (prefix.isEmpty()) {
       prefix = Optional.of(findMostRecentPrefixForWatermark(watermark, bucket, tld, gcsUtils));
     }
     logger.atInfo().log("Verifying readiness to upload the RDE deposit.");

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

@@ -161,7 +161,7 @@ final class RydeEncryption {
               .filter(ciphertext -> ciphertext.getKeyID() == privateKey.getKeyID())
               .findAny();
       // If we can't find one with our key ID, then we can't decrypt the file!
-      if (!cyphertext.isPresent()) {
+      if (cyphertext.isEmpty()) {
         String keyIds =
             PgpUtils.stream(ciphertextList, PGPPublicKeyEncryptedData.class)
                 .map(ciphertext -> Long.toHexString(ciphertext.getKeyID()))

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

@@ -104,7 +104,7 @@ public final class CopyDetailReportsAction implements Runnable {
       // TODO(larryruili): Determine a safer way of enforcing this.
       String registrarId = Iterables.get(Splitter.on('_').split(detailReportName), 3);
       Optional<Registrar> registrar = Registrar.loadByRegistrarId(registrarId);
-      if (!registrar.isPresent()) {
+      if (registrar.isEmpty()) {
         logger.atWarning().log(
             "Registrar %s not found in database for file '%s'.", registrar, detailReportName);
         continue;

core/src/main/java/google/registry/request/RequestHandler.java

@@ -132,7 +132,7 @@ public class RequestHandler<C> {
     }
     String path = req.getRequestURI();
     Optional<Route> route = router.route(path);
-    if (!route.isPresent()) {
+    if (route.isEmpty()) {
       logger.atInfo().log("No action found for: %s", path);
       rsp.sendError(SC_NOT_FOUND);
       return;
@@ -144,7 +144,7 @@ public class RequestHandler<C> {
     }
     Optional<AuthResult> authResult =
         requestAuthenticator.authorize(route.get().action().auth().authSettings(), req);
-    if (!authResult.isPresent()) {
+    if (authResult.isEmpty()) {
       rsp.sendError(SC_FORBIDDEN, "Not authorized");
       return;
     }

core/src/main/java/google/registry/request/auth/AuthenticatedRegistrarAccessor.java

@@ -262,7 +262,7 @@ public class AuthenticatedRegistrarAccessor {
       Lazy<GroupsConnection> lazyGroupsConnection,
       String userEmail,
       Optional<String> gSuiteSupportGroupEmailAddress) {
-    if (!gSuiteSupportGroupEmailAddress.isPresent()) {
+    if (gSuiteSupportGroupEmailAddress.isEmpty()) {
       return false;
     }
     try {
@@ -282,7 +282,7 @@ public class AuthenticatedRegistrarAccessor {
       AuthResult authResult,
       Optional<String> gSuiteSupportGroupEmailAddress,
       Lazy<GroupsConnection> lazyGroupsConnection) {
-    if (!authResult.userAuthInfo().isPresent()) {
+    if (authResult.userAuthInfo().isEmpty()) {
       return false;
     }
 
@@ -300,7 +300,7 @@ public class AuthenticatedRegistrarAccessor {
   /** Returns a map of registrar IDs to roles for all registrars that the user has access to. */
   private static ImmutableSetMultimap<String, Role> createRoleMap(
       AuthResult authResult, boolean isAdmin, String registryAdminRegistrarId) {
-    if (!authResult.userAuthInfo().isPresent()) {
+    if (authResult.userAuthInfo().isEmpty()) {
       return ImmutableSetMultimap.of();
     }
     ImmutableSetMultimap.Builder<String, Role> builder = new ImmutableSetMultimap.Builder<>();

core/src/main/java/google/registry/request/lock/LockHandlerImpl.java

@@ -24,8 +24,8 @@ import com.google.common.collect.ImmutableSortedSet;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.util.concurrent.UncheckedExecutionException;
 import google.registry.model.server.Lock;
-import google.registry.util.AppEngineTimeLimiter;
 import google.registry.util.Clock;
+import google.registry.util.TimeLimiter;
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
@@ -82,7 +82,7 @@ public class LockHandlerImpl implements LockHandler {
     DateTime startTime = clock.nowUtc();
     String sanitizedTld = Strings.emptyToNull(tld);
     try {
-      return AppEngineTimeLimiter.create()
+      return TimeLimiter.create()
           .callWithTimeout(
               new LockingCallable(callable, lockAcquirer, sanitizedTld, leaseLength, lockNames),
               leaseLength.minus(LOCK_TIMEOUT_FUDGE).getMillis(),
@@ -147,7 +147,7 @@ public class LockHandlerImpl implements LockHandler {
       try {
         for (String lockName : lockNames) {
           Optional<Lock> lock = lockAcquirer.acquireLock(lockName, tld, leaseLength);
-          if (!lock.isPresent()) {
+          if (lock.isEmpty()) {
             logger.atInfo().log("Couldn't acquire lock named: %s for TLD %s.", lockName, tld);
             return false;
           }

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

@@ -1,41 +0,0 @@
-// Copyright 2018 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;
-
-import com.google.api.services.appengine.v1.Appengine;
-import dagger.Module;
-import dagger.Provides;
-import google.registry.config.CredentialModule.LocalCredential;
-import google.registry.config.RegistryConfig.Config;
-import google.registry.util.GoogleCredentialsBundle;
-import javax.inject.Singleton;
-
-/** Module providing the instance of {@link Appengine} to access App Engine Admin Api. */
-@Module
-public abstract class AppEngineAdminApiModule {
-
-  @Provides
-  @Singleton
-  public static Appengine provideAppengine(
-      @LocalCredential GoogleCredentialsBundle credentialsBundle,
-      @Config("projectId") String projectId) {
-    return new Appengine.Builder(
-            credentialsBundle.getHttpTransport(),
-            credentialsBundle.getJsonFactory(),
-            credentialsBundle.getHttpRequestInitializer())
-        .setApplicationName(projectId)
-        .build();
-  }
-}

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

@@ -14,6 +14,7 @@
 package google.registry.tools;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static google.registry.model.tld.Tld.Builder.ROID_SUFFIX_PATTERN;
 import static google.registry.model.tld.Tlds.getTlds;
 import static google.registry.util.ListNamingUtils.convertFilePathToName;
@@ -28,10 +29,12 @@ import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.collect.Sets;
 import com.google.common.collect.Sets.SetView;
 import com.google.common.flogger.FluentLogger;
+import google.registry.model.common.TimedTransitionProperty;
 import google.registry.model.tld.Tld;
 import google.registry.model.tld.label.PremiumList;
 import google.registry.model.tld.label.PremiumListDao;
 import google.registry.tools.params.PathParameter;
+import google.registry.util.Clock;
 import google.registry.util.Idn;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
@@ -40,9 +43,11 @@ import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import javax.inject.Inject;
 import javax.inject.Named;
 import org.joda.money.CurrencyUnit;
@@ -90,6 +95,8 @@ public class ConfigureTldCommand extends MutatingCommand {
 
   @Inject ObjectMapper mapper;
 
+  @Inject Clock clock;
+
   @Inject
   @Named("dnsWriterNames")
   Set<String> validDnsWriterNames;
@@ -114,8 +121,11 @@ public class ConfigureTldCommand extends MutatingCommand {
     String name = convertFilePathToName(inputFile);
     Map<String, Object> tldData = new Yaml().load(Files.newBufferedReader(inputFile));
     checkName(name, tldData);
-    checkForMissingFields(tldData);
     Tld oldTld = getTlds().contains(name) ? Tld.get(name) : null;
+
+    checkForMissingFields(
+        Stream.concat(tldData.keySet().stream(), Stream.of("createBillingCostTransitions"))
+            .collect(toImmutableSet()));
     Tld newTld = mapper.readValue(inputFile.toFile(), Tld.class);
     if (oldTld != null) {
       oldTldInBreakGlass = oldTld.getBreakglassMode();
@@ -146,6 +156,14 @@ public class ConfigureTldCommand extends MutatingCommand {
     checkPremiumList(newTld);
     checkDnsWriters(newTld);
     checkCurrency(newTld);
+    // TODO(sarahbot@): Remove this once the createBillingCost field is removed
+    checkArgument(
+        Objects.equals(
+            TimedTransitionProperty.fromValueMap(newTld.getCreateBillingCostTransitions())
+                .getValueAtTime(clock.nowUtc()),
+            newTld.getCreateBillingCost()),
+        "The createBillingCostTransitions map must have the same current cost as the"
+            + " createBillingCost field");
     // bsaEnrollStartTime only exists in DB. Need to carry it over to the updated copy. See Tld.java
     // for more information.
     Optional<DateTime> bsaEnrollTime =
@@ -198,7 +216,7 @@ public class ConfigureTldCommand extends MutatingCommand {
         ROID_SUFFIX_PATTERN.pattern());
   }
 
-  private void checkForMissingFields(Map<String, Object> tldData) {
+  private void checkForMissingFields(ImmutableSet<String> keySet) {
     Set<String> tldFields =
         Arrays.stream(Tld.class.getDeclaredFields())
             .filter(field -> !Modifier.isStatic(field.getModifiers()))
@@ -207,7 +225,7 @@ public class ConfigureTldCommand extends MutatingCommand {
             .collect(Collectors.toSet());
     Set<String> missingFields = new HashSet<>();
     for (String field : tldFields) {
-      if (!tldData.containsKey(field)) {
+      if (!keySet.contains(field)) {
         missingFields.add(field);
       }
     }
@@ -219,7 +237,7 @@ public class ConfigureTldCommand extends MutatingCommand {
 
   private void checkPremiumList(Tld newTld) {
     Optional<String> premiumListName = newTld.getPremiumListName();
-    if (!premiumListName.isPresent()) {
+    if (premiumListName.isEmpty()) {
       return;
     }
     Optional<PremiumList> premiumList = PremiumListDao.getLatestRevision(premiumListName.get());

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

@@ -31,7 +31,7 @@ public final class CreateBulkPricingPackageCommand extends CreateOrUpdateBulkPri
   @Override
   BulkPricingPackage getOldBulkPricingPackage(String tokenString) {
     checkArgument(
-        !BulkPricingPackage.loadByTokenString(tokenString).isPresent(),
+        BulkPricingPackage.loadByTokenString(tokenString).isEmpty(),
         "BulkPricingPackage with token %s already exists",
         tokenString);
     return null;

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

@@ -47,7 +47,7 @@ public class CreatePremiumListCommand extends CreateOrUpdatePremiumListCommand {
     currency = CurrencyUnit.of(currencyUnit);
     name = Strings.isNullOrEmpty(name) ? convertFilePathToName(inputFile) : name;
     checkArgument(
-        !PremiumListDao.getLatestRevision(name).isPresent(),
+        PremiumListDao.getLatestRevision(name).isEmpty(),
         "A premium list already exists by this name");
     if (!override) {
       assertTldExists(

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

@@ -81,9 +81,7 @@ final class CreateRegistrarCommand extends CreateOrUpdateRegistrarCommand
           clientId);
     }
     checkState(
-        !Registrar.loadByRegistrarId(clientId).isPresent(),
-        "Registrar %s already exists",
-        clientId);
+        Registrar.loadByRegistrarId(clientId).isEmpty(), "Registrar %s already exists", clientId);
     List<Registrar> collisions =
         Streams.stream(Registrar.loadAll())
             .filter(registrar -> normalizeRegistrarId(registrar.getRegistrarId()).equals(clientId))

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

@@ -47,8 +47,7 @@ final class CreateReservedListCommand extends CreateOrUpdateReservedListCommand
   @Override
   protected String prompt() throws Exception {
     name = Strings.isNullOrEmpty(name) ? convertFilePathToName(input) : name;
-    checkArgument(
-        !ReservedList.get(name).isPresent(), "A reserved list already exists by this name");
+    checkArgument(ReservedList.get(name).isEmpty(), "A reserved list already exists by this name");
     if (!override) {
       validateListName(name);
     }

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

@@ -28,8 +28,7 @@ public class CreateUserCommand extends CreateOrUpdateUserCommand {
   @Nullable
   @Override
   User getExistingUser(String email) {
-    checkArgument(
-        !UserDao.loadUser(email).isPresent(), "A user with email %s already exists", email);
+    checkArgument(UserDao.loadUser(email).isEmpty(), "A user with email %s already exists", email);
     return null;
   }
 }

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

@@ -18,7 +18,6 @@ import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static com.google.common.collect.Iterables.partition;
 import static com.google.common.collect.Streams.stream;
-import static google.registry.model.domain.token.AllocationToken.TokenType.SINGLE_USE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 
 import com.beust.jcommander.Parameter;
@@ -77,8 +76,8 @@ final class DeleteAllocationTokensCommand extends UpdateOrDeleteAllocationTokens
     // since the query ran. This also filters out per-domain tokens if they're not to be deleted.
     ImmutableSet<VKey<AllocationToken>> tokensToDelete =
         tm().loadByKeys(batch).values().stream()
-            .filter(t -> withDomains || !t.getDomainName().isPresent())
-            .filter(t -> SINGLE_USE.equals(t.getTokenType()))
+            .filter(t -> withDomains || t.getDomainName().isEmpty())
+            .filter(t -> t.getTokenType().isOneTimeUse())
             .filter(t -> !t.isRedeemed())
             .map(AllocationToken::createVKey)
             .collect(toImmutableSet());

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

@@ -102,7 +102,7 @@ public final class DomainLockUtils {
               RegistryLock lock = getByVerificationCode(verificationCode);
 
               checkArgument(
-                  !lock.getLockCompletionTime().isPresent(),
+                  lock.getLockCompletionTime().isEmpty(),
                   "Domain %s is already locked",
                   lock.getDomainName());
 
@@ -129,7 +129,7 @@ public final class DomainLockUtils {
                   DateTime now = tm().getTransactionTime();
                   RegistryLock previousLock = getByVerificationCode(verificationCode);
                   checkArgument(
-                      !previousLock.getUnlockCompletionTime().isPresent(),
+                      previousLock.getUnlockCompletionTime().isEmpty(),
                       "Domain %s is already unlocked",
                       previousLock.getDomainName());
 
@@ -298,7 +298,7 @@ public final class DomainLockUtils {
       checkArgument(
           lock.isLocked(), "Lock object for domain %s is not currently locked", domainName);
       checkArgument(
-          !lock.getUnlockRequestTime().isPresent() || lock.isUnlockRequestExpired(now),
+          lock.getUnlockRequestTime().isEmpty() || lock.isUnlockRequestExpired(now),
           "A pending unlock action already exists for %s",
           domainName);
       checkArgument(

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

@@ -62,7 +62,7 @@ final class GetAllocationTokenCommand implements Command {
       if (loadedTokens.containsKey(token)) {
         AllocationToken loadedToken = loadedTokens.get(token);
         System.out.println(loadedToken.toString());
-        if (!loadedToken.getRedemptionHistoryId().isPresent()) {
+        if (loadedToken.getRedemptionHistoryId().isEmpty()) {
           System.out.printf("Token %s was not redeemed.\n", token);
         } else {
           VKey<Domain> domainKey =

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

@@ -96,7 +96,7 @@ class LoadTestCommand extends ConfirmingCommand implements CommandWithConnection
       errorPrintStream.printf("No such TLD: %s\n", tld);
       return false;
     }
-    if (!Registrar.loadByRegistrarId(clientId).isPresent()) {
+    if (Registrar.loadByRegistrarId(clientId).isEmpty()) {
       errorPrintStream.printf("No such client: %s\n", clientId);
       return false;
     }

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

@@ -55,7 +55,6 @@ import javax.inject.Singleton;
 @Singleton
 @Component(
     modules = {
-      AppEngineAdminApiModule.class,
       AuthModule.class,
       BatchModule.class,
       BigqueryModule.class,

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

@@ -93,8 +93,7 @@ class UnrenewDomainCommand extends ConfirmingCommand {
         continue;
       }
       Optional<Domain> domain = loadByForeignKey(Domain.class, domainName, now);
-      if (!domain.isPresent()
-          || domain.get().getStatusValues().contains(StatusValue.PENDING_DELETE)) {
+      if (domain.isEmpty() || domain.get().getStatusValues().contains(StatusValue.PENDING_DELETE)) {
         domainsDeletingBuilder.add(domainName);
         continue;
       }

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

@@ -52,7 +52,7 @@ import org.joda.time.DateTime;
 final class UpdateAllocationTokensCommand extends UpdateOrDeleteAllocationTokensCommand {
 
   @Parameter(
-      names = {"--allowed_client_ids"},
+      names = {"-c", "--allowed_client_ids"},
       description =
           "Comma-separated list of allowed client IDs. Use the empty string to clear the "
               + "existing list.",
@@ -60,7 +60,7 @@ final class UpdateAllocationTokensCommand extends UpdateOrDeleteAllocationTokens
   private List<String> allowedClientIds;
 
   @Parameter(
-      names = {"--allowed_tlds"},
+      names = {"-t", "--allowed_tlds"},
       description =
           "Comma-separated list of allowed TLDs. Use the empty string to clear the "
               + "existing list.",
@@ -91,7 +91,7 @@ final class UpdateAllocationTokensCommand extends UpdateOrDeleteAllocationTokens
   private Boolean discountPremiums;
 
   @Parameter(
-      names = {"--discount_years"},
+      names = {"-y", "--discount_years"},
       description = "The number of years the discount applies for. Default is 1, max value is 10.")
   private Integer discountYears;
 

core/src/main/java/google/registry/tools/server/CreateGroupsAction.java

@@ -113,11 +113,11 @@ public class CreateGroupsAction implements Runnable {
   }
 
   private Registrar initAndLoadRegistrar() {
-    if (!clientId.isPresent()) {
+    if (clientId.isEmpty()) {
       respondToBadRequest("Error creating Google Groups, missing parameter: clientId");
     }
     Optional<Registrar> registrar = Registrar.loadByRegistrarId(clientId.get());
-    if (!registrar.isPresent()) {
+    if (registrar.isEmpty()) {
       respondToBadRequest(String.format(
           "Error creating Google Groups; could not find registrar with id %s", clientId.get()));
     }

core/src/main/java/google/registry/tools/server/ListObjectsAction.java

@@ -139,7 +139,7 @@ public abstract class ListObjectsAction<T extends ImmutableObject> implements Ru
   private ImmutableSet<String> getFieldsToUse(ImmutableSet<T> objects) {
     // Get the list of fields from the received parameter.
     List<String> fieldsToUse;
-    if ((fields == null) || !fields.isPresent()) {
+    if ((fields == null) || fields.isEmpty()) {
       fieldsToUse = new ArrayList<>();
     } else {
       fieldsToUse = Splitter.on(',').splitToList(fields.get());

core/src/main/java/google/registry/tools/server/RefreshDnsForAllDomainsAction.java

@@ -97,7 +97,7 @@ public class RefreshDnsForAllDomainsAction implements Runnable {
   public void run() {
     assertTldsExist(tlds);
     checkArgument(batchSize > 0, "Must specify a positive number for batch size");
-    Duration smear = tm().transact(this::calculateSmear, TRANSACTION_REPEATABLE_READ);
+    Duration smear = tm().transact(TRANSACTION_REPEATABLE_READ, this::calculateSmear);
 
     ImmutableList<String> domainsBatch;
     @Nullable String lastInPreviousBatch = null;
@@ -105,7 +105,7 @@ public class RefreshDnsForAllDomainsAction implements Runnable {
       Optional<String> lastInPreviousBatchOpt = Optional.ofNullable(lastInPreviousBatch);
       domainsBatch =
           tm().transact(
-                  () -> refreshBatch(lastInPreviousBatchOpt, smear), TRANSACTION_REPEATABLE_READ);
+                  TRANSACTION_REPEATABLE_READ, () -> refreshBatch(lastInPreviousBatchOpt, smear));
       lastInPreviousBatch = domainsBatch.isEmpty() ? null : getLast(domainsBatch);
     } while (domainsBatch.size() == batchSize);
   }

core/src/main/java/google/registry/ui/server/RegistrarFormFields.java

@@ -360,7 +360,7 @@ public final class RegistrarFormFields {
       return ImmutableList.of();
     }
     Optional<List<Map<String, ?>>> contactsAsMaps = CONTACTS_AS_MAPS.extractUntyped(args);
-    if (!contactsAsMaps.isPresent()) {
+    if (contactsAsMaps.isEmpty()) {
       return ImmutableList.of();
     }
     ImmutableList.Builder<RegistrarPoc.Builder> result = new ImmutableList.Builder<>();

core/src/main/java/google/registry/ui/server/console/ConsoleApiAction.java

@@ -35,7 +35,7 @@ public abstract class ConsoleApiAction implements Runnable {
   @Override
   public final void run() {
     // Shouldn't be even possible because of Auth annotations on the various implementing classes
-    if (!consoleApiParams.authResult().userAuthInfo().get().consoleUser().isPresent()) {
+    if (consoleApiParams.authResult().userAuthInfo().get().consoleUser().isEmpty()) {
       consoleApiParams.response().setStatus(HttpStatusCodes.STATUS_CODE_UNAUTHORIZED);
       return;
     }
@@ -62,7 +62,7 @@ public abstract class ConsoleApiAction implements Runnable {
         Arrays.stream(consoleApiParams.request().getCookies())
             .filter(c -> XsrfTokenManager.X_CSRF_TOKEN.equals(c.getName()))
             .findFirst();
-    if (!maybeCookie.isPresent()
+    if (maybeCookie.isEmpty()
         || !consoleApiParams.xsrfTokenManager().validateToken(maybeCookie.get().getValue())) {
       consoleApiParams.response().setStatus(HttpStatusCodes.STATUS_CODE_UNAUTHORIZED);
       return false;

core/src/main/java/google/registry/ui/server/console/ConsoleDomainGetAction.java

@@ -60,12 +60,12 @@ public class ConsoleDomainGetAction implements JsonGetAction {
 
   @Override
   public void run() {
-    if (!authResult.isAuthenticated() || !authResult.userAuthInfo().isPresent()) {
+    if (!authResult.isAuthenticated() || authResult.userAuthInfo().isEmpty()) {
       response.setStatus(HttpStatusCodes.STATUS_CODE_UNAUTHORIZED);
       return;
     }
     UserAuthInfo authInfo = authResult.userAuthInfo().get();
-    if (!authInfo.consoleUser().isPresent()) {
+    if (authInfo.consoleUser().isEmpty()) {
       response.setStatus(HttpStatusCodes.STATUS_CODE_UNAUTHORIZED);
       return;
     }
@@ -75,7 +75,7 @@ public class ConsoleDomainGetAction implements JsonGetAction {
                 () ->
                     EppResourceUtils.loadByForeignKeyCached(
                         Domain.class, paramDomain, tm().getTransactionTime()));
-    if (!possibleDomain.isPresent()) {
+    if (possibleDomain.isEmpty()) {
       response.setStatus(HttpStatusCodes.STATUS_CODE_NOT_FOUND);
       return;
     }

core/src/main/java/google/registry/ui/server/console/RegistrarsAction.java

@@ -107,7 +107,7 @@ public class RegistrarsAction implements JsonGetAction {
       return;
     }
 
-    if (!registrar.isPresent()) {
+    if (registrar.isEmpty()) {
       response.setStatus(HttpStatusCodes.STATUS_CODE_BAD_REQUEST);
       response.setPayload(gson.toJson("'registrar' parameter is not present"));
       return;
@@ -164,7 +164,7 @@ public class RegistrarsAction implements JsonGetAction {
       tm().transact(
               () -> {
                 checkArgument(
-                    !Registrar.loadByRegistrarId(registrar.getRegistrarId()).isPresent(),
+                    Registrar.loadByRegistrarId(registrar.getRegistrarId()).isEmpty(),
                     "Registrar with registrarId %s already exists",
                     registrar.getRegistrarId());
                 tm().putAll(registrar, contact);

core/src/main/java/google/registry/ui/server/console/settings/ContactAction.java

@@ -109,7 +109,7 @@ public class ContactAction implements JsonGetAction {
       return;
     }
 
-    if (!contacts.isPresent()) {
+    if (contacts.isEmpty()) {
       response.setStatus(HttpStatusCodes.STATUS_CODE_BAD_REQUEST);
       response.setPayload(gson.toJson("Contacts parameter is not present"));
       return;

core/src/main/java/google/registry/ui/server/console/settings/SecurityAction.java

@@ -77,7 +77,7 @@ public class SecurityAction implements JsonGetAction {
       return;
     }
 
-    if (!registrar.isPresent()) {
+    if (registrar.isEmpty()) {
       response.setStatus(HttpStatusCodes.STATUS_CODE_BAD_REQUEST);
       response.setPayload(gson.toJson("'registrar' parameter is not present"));
       return;

core/src/main/java/google/registry/ui/server/console/settings/WhoisRegistrarFieldsAction.java

@@ -69,7 +69,7 @@ public class WhoisRegistrarFieldsAction implements JsonGetAction {
 
   @Override
   public void run() {
-    if (!registrar.isPresent()) {
+    if (registrar.isEmpty()) {
       response.setStatus(HttpStatusCodes.STATUS_CODE_BAD_REQUEST);
       response.setPayload(gson.toJson("'registrar' parameter is not present"));
       return;

core/src/main/java/google/registry/ui/server/registrar/ConsoleRegistrarCreatorAction.java

@@ -233,7 +233,7 @@ public final class ConsoleRegistrarCreatorAction extends HtmlAction {
       tm().transact(
               () -> {
                 checkState(
-                    !Registrar.loadByRegistrarId(registrar.getRegistrarId()).isPresent(),
+                    Registrar.loadByRegistrarId(registrar.getRegistrarId()).isEmpty(),
                     "Registrar with client ID %s already exists",
                     registrar.getRegistrarId());
                 tm().putAll(registrar, contact);

core/src/main/java/google/registry/ui/server/registrar/HtmlAction.java

@@ -67,7 +67,7 @@ public abstract class HtmlAction implements Runnable {
     response.setHeader(X_FRAME_OPTIONS, "SAMEORIGIN"); // Disallow iframing.
     response.setHeader("X-Ui-Compatible", "IE=edge"); // Ask IE not to be silly.
 
-    if (!authResult.userAuthInfo().isPresent()) {
+    if (authResult.userAuthInfo().isEmpty()) {
       response.setStatus(SC_MOVED_TEMPORARILY);
       String location;
       try {

core/src/main/java/google/registry/ui/server/registrar/RegistrarSettingsAction.java

@@ -383,7 +383,7 @@ public class RegistrarSettingsAction implements Runnable, JsonActionRunner.JsonA
    */
   private boolean validateCertificate(
       Optional<String> existingCertificate, String certificateString) {
-    if (!existingCertificate.isPresent() || !existingCertificate.get().equals(certificateString)) {
+    if (existingCertificate.isEmpty() || !existingCertificate.get().equals(certificateString)) {
       try {
         certificateChecker.validateCertificate(certificateString);
       } catch (InsecureCertificateException e) {
@@ -514,7 +514,7 @@ public class RegistrarSettingsAction implements Runnable, JsonActionRunner.JsonA
     }
     // If there was a domain WHOIS abuse contact in the old set, the new set must have one.
     if (getDomainWhoisVisibleAbuseContact(existingContacts).isPresent()
-        && !domainWhoisAbuseContact.isPresent()) {
+        && domainWhoisAbuseContact.isEmpty()) {
       throw new ContactRequirementException(
           "An abuse contact visible in domain WHOIS query must be designated");
     }
@@ -569,7 +569,7 @@ public class RegistrarSettingsAction implements Runnable, JsonActionRunner.JsonA
                           updatedContact ->
                               updatedContact.getEmailAddress().equals(contact.getEmailAddress()))
                       .findFirst();
-              if (!updatedContactOptional.isPresent()) {
+              if (updatedContactOptional.isEmpty()) {
                 throw new FormException(
                     String.format(
                         "Cannot delete the contact %s that has registry lock enabled",

core/src/main/java/google/registry/ui/server/registrar/RegistryLockGetAction.java

@@ -210,11 +210,11 @@ public final class RegistryLockGetAction implements JsonGetAction {
         .put(DOMAIN_NAME_PARAM, lock.getDomainName())
         .put(LOCKED_TIME_PARAM, lock.getLockCompletionTime().map(DateTime::toString).orElse(""))
         .put(LOCKED_BY_PARAM, lock.isSuperuser() ? "admin" : lock.getRegistrarPocId())
-        .put(IS_LOCK_PENDING_PARAM, !lock.getLockCompletionTime().isPresent())
+        .put(IS_LOCK_PENDING_PARAM, lock.getLockCompletionTime().isEmpty())
         .put(
             IS_UNLOCK_PENDING_PARAM,
             lock.getUnlockRequestTime().isPresent()
-                && !lock.getUnlockCompletionTime().isPresent()
+                && lock.getUnlockCompletionTime().isEmpty()
                 && !lock.isUnlockRequestExpired(now))
         .put(USER_CAN_UNLOCK_PARAM, isAdmin || !lock.isSuperuser())
         .build();

core/src/main/java/google/registry/whois/DomainLookupCommand.java

@@ -62,7 +62,7 @@ public class DomainLookupCommand implements WhoisCommand {
     Optional<InternetDomainName> tld = findTldForName(domainName);
     // Google Registry Policy: Do not return records under TLDs for which we're not
     // authoritative.
-    if (!tld.isPresent() || !getTlds().contains(tld.get().toString())) {
+    if (tld.isEmpty() || !getTlds().contains(tld.get().toString())) {
       throw new WhoisException(now, SC_NOT_FOUND, ERROR_PREFIX + " not found.");
     }
     // Include `getResponse` and `isBlockedByBsa` in one transaction to reduce latency.

core/src/main/java/google/registry/whois/DomainWhoisResponse.java

@@ -146,7 +146,7 @@ final class DomainWhoisResponse extends WhoisResponseImpl {
     /** Emit the contact entry of the given type. */
     DomainEmitter emitContact(
         String contactType, Optional<VKey<Contact>> contact, boolean preferUnicode) {
-      if (!contact.isPresent()) {
+      if (contact.isEmpty()) {
         return this;
       }
       // If we refer to a contact that doesn't exist, that's a bug. It means referential integrity

core/src/main/java/google/registry/whois/WhoisReader.java

@@ -196,7 +196,7 @@ class WhoisReader {
         // We don't know at this point whether we have a domain name or a host name. We have to
         // search through our configured TLDs to see if there's one that prefixes the name.
         Optional<InternetDomainName> tld = findTldForName(targetName);
-        if (!tld.isPresent()) {
+        if (tld.isEmpty()) {
           // This target is not under any configured TLD, so just try it as a registrar name.
           logger.atInfo().log("Attempting registrar lookup using %s as a registrar.", arg1);
           return commandFactory.registrarLookup(arg1);

core/src/test/java/google/registry/beam/TestPipelineExtension.java

@@ -285,7 +285,7 @@ public class TestPipelineExtension extends Pipeline
     options.as(ApplicationNameOptions.class).setAppName(getAppName(context));
 
     // if the enforcement level has not been set by the user do auto-inference
-    if (!enforcement.isPresent()) {
+    if (enforcement.isEmpty()) {
       final boolean isCrashingRunner = CrashingRunner.class.isAssignableFrom(options.getRunner());
 
       checkState(

core/src/test/java/google/registry/beam/billing/BillingEventTest.java

@@ -16,12 +16,14 @@ package google.registry.beam.billing;
 
 import static com.google.common.truth.Truth.assertThat;
 
+import google.registry.beam.billing.BillingEvent.BillingEventCoder;
 import google.registry.beam.billing.BillingEvent.InvoiceGroupingKey;
 import google.registry.beam.billing.BillingEvent.InvoiceGroupingKey.InvoiceGroupingKeyCoder;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import org.apache.beam.sdk.coders.NullableCoder;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.junit.jupiter.api.BeforeEach;
@@ -120,13 +122,22 @@ class BillingEventTest {
   @Test
   void testInvoiceGroupingKeyCoder_deterministicSerialization() throws IOException {
     InvoiceGroupingKey invoiceKey = event.getInvoiceGroupingKey();
-    InvoiceGroupingKeyCoder coder = new InvoiceGroupingKeyCoder();
+    InvoiceGroupingKeyCoder coder = InvoiceGroupingKeyCoder.of();
     ByteArrayOutputStream outStream = new ByteArrayOutputStream();
     coder.encode(invoiceKey, outStream);
     InputStream inStream = new ByteArrayInputStream(outStream.toByteArray());
     assertThat(coder.decode(inStream)).isEqualTo(invoiceKey);
   }
 
+  @Test
+  void testBillingEventCoder_deterministicSerialization() throws IOException {
+    NullableCoder<BillingEvent> coder = BillingEventCoder.ofNullable();
+    ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+    coder.encode(event, outStream);
+    InputStream inStream = new ByteArrayInputStream(outStream.toByteArray());
+    assertThat(coder.decode(inStream)).isEqualTo(event);
+  }
+
   @Test
   void testGetDetailReportHeader() {
     assertThat(BillingEvent.getHeader())

core/src/test/java/google/registry/beam/billing/InvoicingPipelineTest.java

@@ -36,6 +36,7 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.testing.TestLogHandler;
 import google.registry.beam.TestPipelineExtension;
+import google.registry.beam.billing.BillingEvent.BillingEventCoder;
 import google.registry.model.billing.BillingBase.Flag;
 import google.registry.model.billing.BillingBase.Reason;
 import google.registry.model.billing.BillingCancellation;
@@ -51,13 +52,13 @@ import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationT
 import google.registry.testing.FakeClock;
 import google.registry.util.ResourceUtils;
 import java.io.File;
+import java.io.Serial;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.logging.Logger;
-import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.transforms.Create;
@@ -251,9 +252,7 @@ class InvoicingPipelineTest {
     options.setYearMonth(YEAR_MONTH);
     options.setInvoiceFilePrefix(INVOICE_FILE_PREFIX);
     billingEvents =
-        pipeline.apply(
-            Create.of(INPUT_EVENTS)
-                .withCoder(SerializableCoder.of(google.registry.beam.billing.BillingEvent.class)));
+        pipeline.apply(Create.of(INPUT_EVENTS).withCoder(BillingEventCoder.ofNullable()));
   }
 
   @Test
@@ -346,26 +345,26 @@ class InvoicingPipelineTest {
 
   @Test
   void testSuccess_makeCloudSqlQuery() throws Exception {
-    // Pipeline must be run due to the TestPipelineExtension
+    // The Pipeline must run due to TestPipelineExtension's checks.
     pipeline.run().waitUntilFinish();
     // Test that comments are removed from the .sql file correctly
     assertThat(InvoicingPipeline.makeCloudSqlQuery("2017-10"))
         .isEqualTo(
-            '\n'
-                + "SELECT b, r FROM BillingEvent b\n"
-                + "JOIN Registrar r ON b.clientId = r.registrarId\n"
-                + "JOIN Domain d ON b.domainRepoId = d.repoId\n"
-                + "JOIN Tld t ON t.tldStr = d.tld\n"
-                + "LEFT JOIN BillingCancellation c ON b.id = c.billingEvent\n"
-                + "LEFT JOIN BillingCancellation cr ON b.cancellationMatchingBillingEvent ="
-                + " cr.billingRecurrence\n"
-                + "WHERE r.billingAccountMap IS NOT NULL\n"
-                + "AND r.type = 'REAL'\n"
-                + "AND t.invoicingEnabled IS TRUE\n"
-                + "AND b.billingTime BETWEEN CAST('2017-10-01' AS timestamp) AND CAST('2017-11-01'"
-                + " AS timestamp)\n"
-                + "AND c.id IS NULL\n"
-                + "AND cr.id IS NULL\n");
+            """
+
+      SELECT b, r FROM BillingEvent b
+      JOIN Registrar r ON b.clientId = r.registrarId
+      JOIN Domain d ON b.domainRepoId = d.repoId
+      JOIN Tld t ON t.tldStr = d.tld
+      LEFT JOIN BillingCancellation c ON b.id = c.billingEvent
+      LEFT JOIN BillingCancellation cr ON b.cancellationMatchingBillingEvent = cr.billingRecurrence
+      WHERE r.billingAccountMap IS NOT NULL
+      AND r.type = 'REAL'
+      AND t.invoicingEnabled IS TRUE
+      AND b.billingTime BETWEEN CAST('2017-10-01' AS timestamp) AND CAST('2017-11-01' AS timestamp)
+      AND c.id IS NULL
+      AND cr.id IS NULL
+      """);
   }
 
   /** Returns the text contents of a file under the beamBucket/results directory. */
@@ -604,31 +603,33 @@ class InvoicingPipelineTest {
           PCollection<google.registry.beam.billing.BillingEvent>,
           PCollection<google.registry.beam.billing.BillingEvent>> {
 
-    private static final long serialVersionUID = 2695033474967615250L;
+    @Serial private static final long serialVersionUID = 2695033474967615250L;
 
     @Override
     public PCollection<google.registry.beam.billing.BillingEvent> expand(
         PCollection<google.registry.beam.billing.BillingEvent> input) {
-      return input.apply(
-          "Map to invoicing key",
-          MapElements.into(TypeDescriptor.of(google.registry.beam.billing.BillingEvent.class))
-              .via(
-                  billingEvent ->
-                      google.registry.beam.billing.BillingEvent.create(
-                          billingEvent.id(),
-                          billingEvent.billingTime(),
-                          billingEvent.eventTime(),
-                          billingEvent.registrarId(),
-                          billingEvent.billingId(),
-                          billingEvent.poNumber(),
-                          billingEvent.tld(),
-                          billingEvent.action(),
-                          billingEvent.domain(),
-                          "REPO-ID",
-                          billingEvent.years(),
-                          billingEvent.currency(),
-                          billingEvent.amount(),
-                          billingEvent.flags())));
+      return input
+          .apply(
+              "Map to invoicing key",
+              MapElements.into(TypeDescriptor.of(google.registry.beam.billing.BillingEvent.class))
+                  .via(
+                      billingEvent ->
+                          google.registry.beam.billing.BillingEvent.create(
+                              billingEvent.id(),
+                              billingEvent.billingTime(),
+                              billingEvent.eventTime(),
+                              billingEvent.registrarId(),
+                              billingEvent.billingId(),
+                              billingEvent.poNumber(),
+                              billingEvent.tld(),
+                              billingEvent.action(),
+                              billingEvent.domain(),
+                              "REPO-ID",
+                              billingEvent.years(),
+                              billingEvent.currency(),
+                              billingEvent.amount(),
+                              billingEvent.flags())))
+          .setCoder(BillingEventCoder.ofNullable());
     }
   }
 }

core/src/test/java/google/registry/bsa/UploadBsaUnavailableDomainsActionTest.java

@@ -66,8 +66,6 @@ public class UploadBsaUnavailableDomainsActionTest {
 
   private final GcsUtils gcsUtils = new GcsUtils(LocalStorageHelper.getOptions());
 
-
-
   private final FakeResponse response = new FakeResponse();
 
   @BeforeEach

core/src/test/java/google/registry/flows/domain/DomainCheckFlowTest.java

@@ -19,6 +19,7 @@ import static google.registry.model.billing.BillingBase.RenewalPriceBehavior.DEF
 import static google.registry.model.billing.BillingBase.RenewalPriceBehavior.NONPREMIUM;
 import static google.registry.model.billing.BillingBase.RenewalPriceBehavior.SPECIFIED;
 import static google.registry.model.domain.token.AllocationToken.TokenType.DEFAULT_PROMO;
+import static google.registry.model.domain.token.AllocationToken.TokenType.REGISTER_BSA;
 import static google.registry.model.domain.token.AllocationToken.TokenType.SINGLE_USE;
 import static google.registry.model.domain.token.AllocationToken.TokenType.UNLIMITED_USE;
 import static google.registry.model.eppoutput.CheckData.DomainCheck.create;
@@ -189,6 +190,40 @@ class DomainCheckFlowTest extends ResourceCheckFlowTestCase<DomainCheckFlow, Dom
         create(true, "example3.tld", null));
   }
 
+  @Test
+  void testSuccess_bsaBlocked_createAllowedWithToken() throws Exception {
+    persistBsaLabel("example1");
+    setEppInput("domain_check_allocationtoken.xml");
+    persistResource(
+        new AllocationToken.Builder()
+            .setToken("abc123")
+            .setTokenType(REGISTER_BSA)
+            .setDomainName("example1.tld")
+            .build());
+    doCheckTest(
+        create(true, "example1.tld", null),
+        create(false, "example2.tld", "Alloc token invalid for domain"),
+        create(false, "reserved.tld", "Reserved"),
+        create(false, "specificuse.tld", "Reserved; alloc. token required"));
+  }
+
+  @Test
+  void testSuccess_bsaBlocked_withIrrelevantTokenType() throws Exception {
+    persistBsaLabel("example1");
+    setEppInput("domain_check_allocationtoken.xml");
+    persistResource(
+        new AllocationToken.Builder()
+            .setToken("abc123")
+            .setTokenType(SINGLE_USE)
+            .setDomainName("example1.tld")
+            .build());
+    doCheckTest(
+        create(false, "example1.tld", "Blocked by a GlobalBlock service"),
+        create(false, "example2.tld", "Alloc token invalid for domain"),
+        create(false, "reserved.tld", "Reserved"),
+        create(false, "specificuse.tld", "Reserved; alloc. token required"));
+  }
+
   @Test
   void testSuccess_clTridNotSpecified() throws Exception {
     setEppInput("domain_check_no_cltrid.xml");

core/src/test/java/google/registry/flows/domain/DomainCreateFlowTest.java

@@ -29,6 +29,7 @@ import static google.registry.model.billing.BillingBase.RenewalPriceBehavior.SPE
 import static google.registry.model.domain.fee.Fee.FEE_EXTENSION_URIS;
 import static google.registry.model.domain.token.AllocationToken.TokenType.BULK_PRICING;
 import static google.registry.model.domain.token.AllocationToken.TokenType.DEFAULT_PROMO;
+import static google.registry.model.domain.token.AllocationToken.TokenType.REGISTER_BSA;
 import static google.registry.model.domain.token.AllocationToken.TokenType.SINGLE_USE;
 import static google.registry.model.domain.token.AllocationToken.TokenType.UNLIMITED_USE;
 import static google.registry.model.eppcommon.EppXmlTransformer.marshal;
@@ -255,6 +256,14 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
     persistClaimsList(ImmutableMap.of("example-one", CLAIMS_KEY, "test-validate", CLAIMS_KEY));
   }
 
+  private void enrollTldInBsa() {
+    persistResource(
+        Tld.get("tld")
+            .asBuilder()
+            .setBsaEnrollStartTime(Optional.of(clock.nowUtc().minusSeconds(1)))
+            .build());
+  }
+
   /**
    * Create host and contact entries for testing.
    *
@@ -2593,12 +2602,92 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
   }
 
   @Test
-  void testFailure_blockedByBsa() throws Exception {
+  void testSuccess_blockedByBsa_hasRegisterBsaToken() throws Exception {
+    enrollTldInBsa();
+    allocationToken =
+        persistResource(
+            new AllocationToken.Builder()
+                .setToken("abc123")
+                .setTokenType(REGISTER_BSA)
+                .setDomainName("example.tld")
+                .build());
+    persistBsaLabel("example");
+    persistContactsAndHosts();
+    setEppInput(
+        "domain_create_allocationtoken.xml",
+        ImmutableMap.of("DOMAIN", "example.tld", "YEARS", "2"));
+    runFlow();
+    assertSuccessfulCreate("tld", ImmutableSet.of(), allocationToken);
+  }
+
+  @Test
+  void testSuccess_blockedByBsa_reservedDomain_viaAllocationTokenExtension() throws Exception {
+    enrollTldInBsa();
+    allocationToken =
+        persistResource(
+            new AllocationToken.Builder()
+                .setToken("abc123")
+                .setTokenType(REGISTER_BSA)
+                .setDomainName("resdom.tld")
+                .build());
+    persistBsaLabel("resdom");
+    setEppInput(
+        "domain_create_allocationtoken.xml", ImmutableMap.of("DOMAIN", "resdom.tld", "YEARS", "2"));
+    persistContactsAndHosts();
+    runFlowAssertResponse(
+        loadFile("domain_create_response.xml", ImmutableMap.of("DOMAIN", "resdom.tld")));
+    assertSuccessfulCreate("tld", ImmutableSet.of(RESERVED), allocationToken);
+    assertNoLordn();
+    assertAllocationTokenWasRedeemed("abc123");
+  }
+
+  @Test
+  void testSuccess_blockedByBsa_quietPeriod_skipTldStateCheckWithToken() throws Exception {
+    enrollTldInBsa();
+    AllocationToken token =
+        persistResource(
+            new AllocationToken.Builder()
+                .setToken("abc123")
+                .setTokenType(REGISTER_BSA)
+                .setRegistrationBehavior(RegistrationBehavior.BYPASS_TLD_STATE)
+                .setDomainName("example.tld")
+                .build());
+    persistContactsAndHosts();
+    persistBsaLabel("example");
+    setEppInput(
+        "domain_create_allocationtoken.xml",
+        ImmutableMap.of("DOMAIN", "example.tld", "YEARS", "2"));
     persistResource(
         Tld.get("tld")
             .asBuilder()
-            .setBsaEnrollStartTime(Optional.of(clock.nowUtc().minusSeconds(1)))
+            .setTldStateTransitions(ImmutableSortedMap.of(START_OF_TIME, QUIET_PERIOD))
             .build());
+    runFlow();
+    assertSuccessfulCreate("tld", ImmutableSet.of(), token);
+  }
+
+  @Test
+  void testSuccess_blockedByBsa_anchorTenant() throws Exception {
+    enrollTldInBsa();
+    allocationToken =
+        persistResource(
+            new AllocationToken.Builder()
+                .setToken("abcDEF23456")
+                .setTokenType(REGISTER_BSA)
+                .setDomainName("anchor.tld")
+                .build());
+    setEppInput("domain_create_anchor_allocationtoken.xml");
+    persistContactsAndHosts();
+    persistBsaLabel("anchor");
+    runFlowAssertResponse(loadFile("domain_create_anchor_response.xml"));
+    assertSuccessfulCreate("tld", ImmutableSet.of(ANCHOR_TENANT), allocationToken);
+    assertNoLordn();
+    assertAllocationTokenWasRedeemed("abcDEF23456");
+  }
+
+  @Test
+  void testFailure_blockedByBsa() throws Exception {
+    enrollTldInBsa();
     persistBsaLabel("example");
     persistContactsAndHosts();
     EppException thrown = assertThrows(DomainLabelBlockedByBsaException.class, this::runFlow);
@@ -2619,6 +2708,40 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
         .isEqualTo(loadFile("domain_create_blocked_by_bsa.xml"));
   }
 
+  @Test
+  void testFailure_blockedByBsa_hasWrongToken() throws Exception {
+    enrollTldInBsa();
+    allocationToken =
+        persistResource(
+            new AllocationToken.Builder()
+                .setToken("abc123")
+                .setTokenType(SINGLE_USE)
+                .setRegistrationBehavior(RegistrationBehavior.BYPASS_TLD_STATE)
+                .setDomainName("example.tld")
+                .build());
+    persistBsaLabel("example");
+    persistContactsAndHosts();
+    setEppInput(
+        "domain_create_allocationtoken.xml",
+        ImmutableMap.of("DOMAIN", "example.tld", "YEARS", "2"));
+    EppException thrown = assertThrows(DomainLabelBlockedByBsaException.class, this::runFlow);
+    assertAboutEppExceptions()
+        .that(thrown)
+        .marshalsToXml()
+        .and()
+        .hasMessage("Domain label is blocked by the Brand Safety Alliance");
+    byte[] responseXmlBytes =
+        marshal(
+            EppOutput.create(
+                new EppResponse.Builder()
+                    .setTrid(Trid.create(null, "server-trid"))
+                    .setResult(thrown.getResult())
+                    .build()),
+            ValidationMode.STRICT);
+    assertThat(new String(responseXmlBytes, StandardCharsets.UTF_8))
+        .isEqualTo(loadFile("domain_create_blocked_by_bsa.xml"));
+  }
+
   @Test
   void testFailure_uppercase() {
     doFailingDomainNameTest("Example.tld", BadDomainNameCharacterException.class);

core/src/test/java/google/registry/flows/domain/DomainTransferRequestFlowTest.java

@@ -251,11 +251,6 @@ class DomainTransferRequestFlowTest
                 .build());
   }
 
-  /** Implements the missing Optional.stream function that is added in Java 9. */
-  private static <T> Stream<T> optionalToStream(Optional<T> optional) {
-    return optional.map(Stream::of).orElseGet(Stream::empty);
-  }
-
   private void assertHistoryEntriesContainBillingEventsAndGracePeriods(
       DateTime expectedExpirationTime,
       DateTime implicitTransferTime,
@@ -314,7 +309,7 @@ class DomainTransferRequestFlowTest
     ImmutableSet<BillingBase> expectedBillingBases =
         Streams.concat(
                 Stream.of(losingClientAutorenew, gainingClientAutorenew),
-                optionalToStream(optionalTransferBillingEvent))
+                optionalTransferBillingEvent.stream())
             .collect(toImmutableSet());
     assertBillingEvents(Sets.union(expectedBillingBases, extraBillingBases));
     // Assert that the domain's TransferData server-approve billing events match the above.
@@ -331,8 +326,7 @@ class DomainTransferRequestFlowTest
     // Assert that the full set of server-approve billing events is exactly the extra ones plus
     // the transfer billing event (if present) and the gaining client autorenew.
     ImmutableSet<BillingBase> expectedServeApproveBillingBases =
-        Streams.concat(
-                Stream.of(gainingClientAutorenew), optionalToStream(optionalTransferBillingEvent))
+        Streams.concat(Stream.of(gainingClientAutorenew), optionalTransferBillingEvent.stream())
             .collect(toImmutableSet());
     assertBillingEventsEqual(
         Iterables.filter(