Use stream-collect rather than transformValues in Spec11 collection

As it turns out, using Maps::transformValues does not allow us to change the
resulting map--calling Map::put throws an UnsupportedOperationException. As a
result, we have to do this roundabout stream-collect to do a group-by.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=244852373

java/google/registry/reporting/spec11/PublishSpec11ReportAction.java

@@ -14,7 +14,6 @@
 
 package google.registry.reporting.spec11;
 
-import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static google.registry.reporting.ReportingModule.PARAM_DATE;
 import static google.registry.request.Action.Method.POST;
@@ -25,10 +24,12 @@ import static javax.servlet.http.HttpServletResponse.SC_OK;
 
 import com.google.api.services.dataflow.Dataflow;
 import com.google.api.services.dataflow.model.Job;
-import com.google.common.collect.ImmutableListMultimap;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multimaps;
+import com.google.common.collect.Sets;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
 import com.google.template.soy.parseinfo.SoyTemplateInfo;
@@ -41,11 +42,8 @@ import google.registry.request.Parameter;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import java.io.IOException;
-import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
-import java.util.stream.Collectors;
 import javax.inject.Inject;
 import org.joda.time.LocalDate;
 import org.json.JSONException;
@@ -170,31 +168,37 @@ public class PublishSpec11ReportAction implements Runnable {
   }
 
   private ImmutableSet<RegistrarThreatMatches> getNewMatches(
-      Set<RegistrarThreatMatches> previousMatchesSet,
+      ImmutableSet<RegistrarThreatMatches> previousMatchesSet,
-      Set<RegistrarThreatMatches> currentMatchesSet) {
+      ImmutableSet<RegistrarThreatMatches> currentMatchesSet) {
+    ImmutableMap<String, ImmutableSet<ThreatMatch>> previousMatchesByEmail =
+        groupByKeyAndFlatMap(previousMatchesSet);
+    ImmutableMap<String, ImmutableSet<ThreatMatch>> currentMatchesByEmail =
+        groupByKeyAndFlatMap(currentMatchesSet);
+    ImmutableSet.Builder<RegistrarThreatMatches> resultsBuilder = ImmutableSet.builder();
+    for (String email : currentMatchesByEmail.keySet()) {
+      // Only include matches in the result if they're non-empty
+      Set<ThreatMatch> difference =
+          Sets.difference(
+              currentMatchesByEmail.get(email),
+              previousMatchesByEmail.getOrDefault(email, ImmutableSet.of()));
+      if (!difference.isEmpty()) {
+        resultsBuilder.add(RegistrarThreatMatches.create(email, ImmutableList.copyOf(difference)));
+      }
+    }
+    return resultsBuilder.build();
+  }
+
+  private ImmutableMap<String, ImmutableSet<ThreatMatch>> groupByKeyAndFlatMap(
+      ImmutableSet<RegistrarThreatMatches> registrarThreatMatches) {
     // Group by email address then flat-map all of the ThreatMatch objects together
-    ImmutableListMultimap<String, RegistrarThreatMatches> currentMatchesByEmail =
+    return ImmutableMap.copyOf(
-        Multimaps.index(currentMatchesSet, RegistrarThreatMatches::registrarEmailAddress);
-    Map<String, List<ThreatMatch>> currentMatchMap =
         Maps.transformValues(
-            currentMatchesByEmail.asMap(),
+            Multimaps.index(registrarThreatMatches, RegistrarThreatMatches::registrarEmailAddress)
+                .asMap(),
             registrarThreatMatchesCollection ->
                 registrarThreatMatchesCollection.stream()
                     .flatMap(matches -> matches.threatMatches().stream())
-                    .collect(toImmutableList()));
+                    .collect(toImmutableSet())));
-    previousMatchesSet.forEach(
-        previousMatches ->
-            currentMatchMap.computeIfPresent(
-                previousMatches.registrarEmailAddress(),
-                (email, currentMatches) ->
-                    currentMatches.stream()
-                        .filter(
-                            currentMatch -> !previousMatches.threatMatches().contains(currentMatch))
-                        .collect(Collectors.toList())));
-    return currentMatchMap.entrySet().stream()
-        .filter(entry -> !entry.getValue().isEmpty())
-        .map(entry -> RegistrarThreatMatches.create(entry.getKey(), entry.getValue()))
-        .collect(toImmutableSet());
   }
 
   private boolean shouldSendMonthlySpec11Email() {

javatests/google/registry/reporting/spec11/PublishSpec11ReportActionTest.java

@@ -199,6 +199,24 @@ public class PublishSpec11ReportActionTest {
     verifyNoMoreInteractions(emailUtils);
   }
 
+  @Test
+  public void testJobDone_noDifferentResults() throws Exception {
+    LocalDate yesterday = date.minusDays(1);
+    when(parser.getPreviousDateWithMatches(date)).thenReturn(Optional.of(yesterday));
+    when(parser.getRegistrarThreatMatches(date)).thenReturn(sampleThreatMatches());
+    when(parser.getRegistrarThreatMatches(yesterday)).thenReturn(sampleThreatMatches());
+    expectedJob.setCurrentState("JOB_STATE_DONE");
+    publishAction.run();
+    assertThat(response.getStatus()).isEqualTo(SC_OK);
+    verify(emailUtils)
+        .emailSpec11Reports(
+            date,
+            Spec11EmailSoyInfo.DAILY_SPEC_11_EMAIL,
+            "Super Cool Registry Daily Threat Detector [2018-06-05]",
+            ImmutableSet.of());
+    verifyNoMoreInteractions(emailUtils);
+  }
+
   @Test
   public void testJobDone_failsDueToNoPreviousResults() {
     when(parser.getPreviousDateWithMatches(date)).thenReturn(Optional.empty());