Fix StreamingDataflowWorkerTest, new tag is added

Author: stankiewicz

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

@@ -1807,6 +1807,7 @@ public void testMergeWindows() throws Exception {
     String timerTagPrefix = "/s" + window + "+0";
     ByteString bufferTag = ByteString.copyFromUtf8(window + "+ubuf");
     ByteString paneInfoTag = ByteString.copyFromUtf8(window + "+upaneInfo");
+    ByteString pipelineMetadataTag = ByteString.copyFromUtf8(window + "+upipelineMetadata");
     String watermarkDataHoldTag = window + "+uhold";
     String watermarkExtraHoldTag = window + "+uextra";
     String stateFamily = "MergeWindows";
@@ -1946,11 +1947,20 @@ public void testMergeWindows() throws Exception {
     assertThat(
         "" + actualOutput.getValueUpdatesList(),
         actualOutput.getValueUpdatesList(),
-        Matchers.contains(
+        Matchers.containsInAnyOrder(
             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(pipelineMetadataTag)
+                    .setStateFamily(stateFamily)
                     .setValue(
                         Windmill.Value.newBuilder()
                             .setTimestamp(Long.MAX_VALUE)
@@ -2097,6 +2107,7 @@ public void testMergeWindowsCaching() throws Exception {
     String timerTagPrefix = "/s" + window + "+0";
     ByteString bufferTag = ByteString.copyFromUtf8(window + "+ubuf");
     ByteString paneInfoTag = ByteString.copyFromUtf8(window + "+upaneInfo");
+    ByteString pipelineMetadataTag = ByteString.copyFromUtf8(window + "+upipelineMetadata");
     String watermarkDataHoldTag = window + "+uhold";
     String watermarkExtraHoldTag = window + "+uextra";
     String stateFamily = "MergeWindows";
@@ -2236,11 +2247,20 @@ public void testMergeWindowsCaching() throws Exception {
     assertThat(
         "" + actualOutput.getValueUpdatesList(),
         actualOutput.getValueUpdatesList(),
-        Matchers.contains(
+        Matchers.containsInAnyOrder(
             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(pipelineMetadataTag)
+                    .setStateFamily(stateFamily)
                     .setValue(
                         Windmill.Value.newBuilder()
                             .setTimestamp(Long.MAX_VALUE)