Skip poll messages on deletions for configured registrars (#2613)

See b/379331882 for more details

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

@@ -1247,4 +1247,15 @@ class DomainDeleteFlowTest extends ResourceFlowTestCase<DomainDeleteFlow, Domain
     clock.advanceOneMilli();
     runFlowAssertResponse(loadFile("domain_delete_response_fee_free_grace.xml"));
   }
+
+  @Test
+  void testSuccess_skipsPollMessage_whenConfigured() throws Exception {
+    setUpSuccessfulTest();
+    domain =
+        persistResource(
+            domain.asBuilder().setPersistedCurrentSponsorRegistrarId("NewRegistrar").build());
+    setRegistrarIdForFlow("NewRegistrar");
+    runFlowAssertResponse(loadFile("domain_delete_response_pending.xml"));
+    assertPollMessages();
+  }
 }

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

@@ -74,6 +74,7 @@ import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
 import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.tld.Tld;
+import google.registry.persistence.VKey;
 import google.registry.testing.DatabaseHelper;
 import java.util.Map;
 import java.util.Optional;
@@ -103,12 +104,12 @@ class DomainRestoreRequestFlowTest extends ResourceFlowTestCase<DomainRestoreReq
     setEppInput("domain_update_restore_request.xml", ImmutableMap.of("DOMAIN", "example.tld"));
   }
 
-  void persistPendingDeleteDomain() throws Exception {
+  Domain persistPendingDeleteDomain() throws Exception {
     // The domain is now past what had been its expiration date at the time of deletion.
-    persistPendingDeleteDomain(clock.nowUtc().minusDays(5));
+    return persistPendingDeleteDomain(clock.nowUtc().minusDays(5));
   }
 
-  void persistPendingDeleteDomain(DateTime expirationTime) throws Exception {
+  Domain persistPendingDeleteDomain(DateTime expirationTime) throws Exception {
     Domain domain = persistResource(DatabaseHelper.newDomain(getUniqueIdFromCommand()));
     HistoryEntry historyEntry =
         persistResource(
@@ -118,29 +119,31 @@ class DomainRestoreRequestFlowTest extends ResourceFlowTestCase<DomainRestoreReq
                 .setRegistrarId(domain.getCurrentSponsorRegistrarId())
                 .setDomain(domain)
                 .build());
-    persistResource(
-        domain
-            .asBuilder()
-            .setRegistrationExpirationTime(expirationTime)
-            .setDeletionTime(clock.nowUtc().plusDays(35))
-            .addGracePeriod(
-                GracePeriod.create(
-                    GracePeriodStatus.REDEMPTION,
-                    domain.getRepoId(),
-                    clock.nowUtc().plusDays(1),
-                    "TheRegistrar",
-                    null))
-            .setStatusValues(ImmutableSet.of(StatusValue.PENDING_DELETE))
-            .setDeletePollMessage(
-                persistResource(
-                        new PollMessage.OneTime.Builder()
-                            .setRegistrarId("TheRegistrar")
-                            .setEventTime(clock.nowUtc().plusDays(5))
-                            .setHistoryEntry(historyEntry)
-                            .build())
-                    .createVKey())
-            .build());
+    domain =
+        persistResource(
+            domain
+                .asBuilder()
+                .setRegistrationExpirationTime(expirationTime)
+                .setDeletionTime(clock.nowUtc().plusDays(35))
+                .addGracePeriod(
+                    GracePeriod.create(
+                        GracePeriodStatus.REDEMPTION,
+                        domain.getRepoId(),
+                        clock.nowUtc().plusDays(1),
+                        "TheRegistrar",
+                        null))
+                .setStatusValues(ImmutableSet.of(StatusValue.PENDING_DELETE))
+                .setDeletePollMessage(
+                    persistResource(
+                            new PollMessage.OneTime.Builder()
+                                .setRegistrarId("TheRegistrar")
+                                .setEventTime(clock.nowUtc().plusDays(5))
+                                .setHistoryEntry(historyEntry)
+                                .build())
+                        .createVKey())
+                .build());
     clock.advanceOneMilli();
+    return domain;
   }
 
   @Test
@@ -491,6 +494,15 @@ class DomainRestoreRequestFlowTest extends ResourceFlowTestCase<DomainRestoreReq
         loadFile("domain_update_restore_request_response_premium.xml"));
   }
 
+  @Test
+  void testSuccess_worksWithoutPollMessage() throws Exception {
+    Domain domain = persistPendingDeleteDomain();
+    VKey<PollMessage.OneTime> deletePollMessage = domain.getDeletePollMessage();
+    persistResource(domain.asBuilder().setDeletePollMessage(null).build());
+    DatabaseHelper.deleteByKey(deletePollMessage);
+    runFlowAssertResponse(loadFile("generic_success_response.xml"));
+  }
+
   @Test
   void testFailure_doesNotExist() throws Exception {
     ResourceDoesNotExistException thrown =

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

@@ -1168,6 +1168,10 @@ public final class DatabaseHelper {
     tm().transact(() -> tm().delete(resource));
   }
 
+  public static void deleteByKey(VKey<?> key) {
+    tm().transact(() -> tm().delete(key));
+  }
+
   /** Force the create and update timestamps to get written into the resource. */
   public static <R> R cloneAndSetAutoTimestamps(final R resource) {
     // We have to separate the read and write operation into different transactions otherwise JPA