Add a "ReplaySpecializer" to fix certain replays (#989)

* Add a "ReplaySpecializer" to fix certain replays

Due to the fact that a given entity in either database type can map to
multiple entities in the other database, there are certain replication
scenarios that don't quite work.  Current known examples include:

- propagation of cascading deletes from datastore to SQL
- creation of datastore indexed entities for SQL entities (where indexes are a
  first-class concept)

This change introduces a ReplaySpecializer class, which allows us to declare
static method hooks at the entity class level that define any special
operations that need to be performed before or after replaying a mutation for
any given entity type.

Currently, "before SQL delete" is the only supported hook.  A change to
DomainContent demonstrating how this facility can be used to fix problems in
cascading delete propagation will be sent as a subsequent PR.

* Throw exception on beforeSqlDelete failures

* Changes for review

core/src/test/java/google/registry/backup/ReplayCommitLogsToSqlActionTest.java

@@ -126,6 +126,7 @@ public class ReplayCommitLogsToSqlActionTest {
     action.diffLister.gcsBucket = GCS_BUCKET;
     action.diffLister.executor = newDirectExecutorService();
     RegistryConfig.overrideCloudSqlReplayCommitLogs(true);
+    TestObject.beforeSqlDeleteCallCount = 0;
   }
 
   @Test
@@ -441,6 +442,21 @@ public class ReplayCommitLogsToSqlActionTest {
         .isEqualTo("Can't acquire SQL commit log replay lock, aborting.");
   }
 
+  @Test
+  void testSuccess_deleteSqlCallback() throws Exception {
+    DateTime now = fakeClock.nowUtc();
+    jpaTm().transact(() -> SqlReplayCheckpoint.set(now.minusMinutes(1).minusMillis(1)));
+    saveDiffFile(
+        gcsService,
+        createCheckpoint(now.minusMinutes(1)),
+        CommitLogManifest.create(
+            getBucketKey(1),
+            now.minusMinutes(1),
+            ImmutableSet.of(Key.create(TestObject.create("to delete")))));
+    action.run();
+    assertThat(TestObject.beforeSqlDeleteCallCount).isEqualTo(1);
+  }
+
   private void runAndAssertSuccess(DateTime expectedCheckpointTime) {
     action.run();
     assertThat(response.getStatus()).isEqualTo(SC_OK);

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

@@ -34,6 +34,8 @@ import javax.persistence.Transient;
 @EntityForTesting
 public class TestObject extends ImmutableObject implements DatastoreAndSqlEntity {
 
+  public static int beforeSqlDeleteCallCount;
+
   @Parent @Transient Key<EntityGroupRoot> parent;
 
   @Id @javax.persistence.Id String id;
@@ -68,6 +70,10 @@ public class TestObject extends ImmutableObject implements DatastoreAndSqlEntity
     return instance;
   }
 
+  public static void beforeSqlDelete(VKey<TestObject> key) {
+    beforeSqlDeleteCallCount++;
+  }
+
   /** A test @VirtualEntity model object, which should not be persisted. */
   @Entity
   @VirtualEntity