fix tests related to new state added, fix spotbugs

Author: stankiewicz

runners/core-java/src/main/java/org/apache/beam/runners/core/CombinedMetadataCombiner.java

@@ -29,17 +29,15 @@ public static CombinedMetadataCombiner of() {
     return INSTANCE;
   }
 
-  private final CausedByDrainCombiner causedByDrainCombiner = CausedByDrainCombiner.of();
-
   @Override
   public CombinedMetadata createAccumulator() {
-    return CombinedMetadata.create(causedByDrainCombiner.createAccumulator());
+    return CombinedMetadata.create(CausedByDrainCombiner.of().createAccumulator());
   }
 
   @Override
   public CombinedMetadata addInput(CombinedMetadata accumulator, CombinedMetadata input) {
     return CombinedMetadata.create(
-        causedByDrainCombiner.addInput(accumulator.causedByDrain(), input.causedByDrain()));
+        CausedByDrainCombiner.of().addInput(accumulator.causedByDrain(), input.causedByDrain()));
   }
 
   @Override

runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java

@@ -1953,20 +1953,11 @@ public void testMergeWindows() throws Exception {
     assertThat(
         "" + actualOutput.getValueUpdatesList(),
         actualOutput.getValueUpdatesList(),
-        Matchers.containsInAnyOrder(
+        Matchers.contains(
             Matchers.equalTo(
                 Windmill.TagValue.newBuilder()
                     .setTag(paneInfoTag)
                     .setStateFamily(stateFamily)
-                    .setValue(
-                        Windmill.Value.newBuilder()
-                            .setTimestamp(Long.MAX_VALUE)
-                            .setData(ByteString.EMPTY))
-                    .build()),
-            Matchers.equalTo(
-                Windmill.TagValue.newBuilder()
-                    .setTag(combinedMetadataTag)
-                    .setStateFamily(stateFamily)
                     .setValue(
                         Windmill.Value.newBuilder()
                             .setTimestamp(Long.MAX_VALUE)
@@ -1976,12 +1967,18 @@ public void testMergeWindows() throws Exception {
     assertThat(
         "" + actualOutput.getBagUpdatesList(),
         actualOutput.getBagUpdatesList(),
-        Matchers.contains(
+        Matchers.containsInAnyOrder(
             Matchers.equalTo(
                 Windmill.TagBag.newBuilder()
                     .setTag(bufferTag)
                     .setStateFamily(stateFamily)
                     .setDeleteAll(true)
+                    .build()),
+            Matchers.equalTo(
+                Windmill.TagBag.newBuilder()
+                    .setTag(combinedMetadataTag)
+                    .setStateFamily(stateFamily)
+                    .setDeleteAll(true)
                     .build())));
 
     verifyHolds(
@@ -2260,20 +2257,11 @@ public void testMergeWindowsCaching() throws Exception {
     assertThat(
         "" + actualOutput.getValueUpdatesList(),
         actualOutput.getValueUpdatesList(),
-        Matchers.containsInAnyOrder(
+        Matchers.contains(
             Matchers.equalTo(
                 Windmill.TagValue.newBuilder()
                     .setTag(paneInfoTag)
                     .setStateFamily(stateFamily)
-                    .setValue(
-                        Windmill.Value.newBuilder()
-                            .setTimestamp(Long.MAX_VALUE)
-                            .setData(ByteString.EMPTY))
-                    .build()),
-            Matchers.equalTo(
-                Windmill.TagValue.newBuilder()
-                    .setTag(combinedMetadataTag)
-                    .setStateFamily(stateFamily)
                     .setValue(
                         Windmill.Value.newBuilder()
                             .setTimestamp(Long.MAX_VALUE)
@@ -2283,12 +2271,18 @@ public void testMergeWindowsCaching() throws Exception {
     assertThat(
         "" + actualOutput.getBagUpdatesList(),
         actualOutput.getBagUpdatesList(),
-        Matchers.contains(
+        Matchers.containsInAnyOrder(
             Matchers.equalTo(
                 Windmill.TagBag.newBuilder()
                     .setTag(bufferTag)
                     .setStateFamily(stateFamily)
                     .setDeleteAll(true)
+                    .build()),
+            Matchers.equalTo(
+                Windmill.TagBag.newBuilder()
+                    .setTag(combinedMetadataTag)
+                    .setStateFamily(stateFamily)
+                    .setDeleteAll(true)
                     .build())));
 
     verifyHolds(