Merge remote-tracking branch 'origin/master' into test_vendored_grpc

buildSrc/src/main/groovy/org/apache/beam/gradle/GrpcVendoring_1_36_0.groovy

@@ -69,8 +69,6 @@ class GrpcVendoring_1_36_0 {
   static List<String> runtimeDependencies() {
     return [
       'com.google.errorprone:error_prone_annotations:2.4.0',
-      'commons-logging:commons-logging:1.2',
-      'org.slf4j:slf4j-api:1.7.30',
       // TODO(BEAM-9288): Enable relocation for conscrypt
       "org.conscrypt:conscrypt-openjdk-uber:$conscrypt_version"
     ]
@@ -158,8 +156,6 @@ class GrpcVendoring_1_36_0 {
       "javax/annotation/**",
       "junit/**",
       "module-info.class",
-      "org/apache/commons/logging/**",
-      "org/apache/log/**",
       "org/checkerframework/**",
       "org/codehaus/mojo/animal_sniffer/**",
       "org/conscrypt/**",
@@ -169,7 +165,6 @@ class GrpcVendoring_1_36_0 {
       "org/junit/**",
       "org/mockito/**",
       "org/objenesis/**",
-      "org/slf4j/**",
     ]
   }
 

runners/core-java/src/main/java/org/apache/beam/runners/core/metrics/ShortIdMap.java

@@ -30,6 +30,9 @@ public class ShortIdMap {
 
   public synchronized String getOrCreateShortId(MonitoringInfo info) {
     Preconditions.checkNotNull(info);
+    Preconditions.checkArgument(info.getPayload().isEmpty());
+    Preconditions.checkArgument(!info.hasStartTime());
+
     String shortId = monitoringInfoMap.inverse().get(info);
     if (shortId == null) {
       shortId = "metric" + counter++;

runners/core-java/src/test/java/org/apache/beam/runners/core/metrics/ShortIdMapTest.java

@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.core.metrics;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.beam.model.pipeline.v1.MetricsApi;
+import org.apache.beam.sdk.values.KV;
+import org.junit.Test;
+
+public class ShortIdMapTest {
+
+  @Test
+  public void testShortIdAssignment() throws Exception {
+    ShortIdMap shortIdMap = new ShortIdMap();
+    List<KV<String, MetricsApi.MonitoringInfo>> testCases = new ArrayList<>();
+
+    SimpleMonitoringInfoBuilder builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_DISTRIBUTION_INT64);
+    testCases.add(KV.of("metric0", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.ELEMENT_COUNT);
+    testCases.add(KV.of("metric1", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_DISTRIBUTION_DOUBLE);
+    testCases.add(KV.of("metric2", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn("TestingSentinelUrn");
+    builder.setType("TestingSentinelType");
+    testCases.add(KV.of("metric3", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.FINISH_BUNDLE_MSECS);
+    testCases.add(KV.of("metric4", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    testCases.add(KV.of("metric5", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    builder.setLabel(MonitoringInfoConstants.Labels.NAME, "metricNumber7");
+    builder.setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "myNamespace");
+    builder.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "myPtransform");
+    testCases.add(KV.of("metric6", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    builder.setLabel(MonitoringInfoConstants.Labels.NAME, "metricNumber8");
+    builder.setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "myNamespace");
+    builder.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "myPtransform");
+    testCases.add(KV.of("metric7", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.API_REQUEST_COUNT);
+    builder.setLabel(MonitoringInfoConstants.Labels.SERVICE, "BigQuery");
+    testCases.add(KV.of("metric8", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.API_REQUEST_COUNT);
+    builder.setLabel(MonitoringInfoConstants.Labels.SERVICE, "Storage");
+    testCases.add(KV.of("metric9", builder.build()));
+
+    // Validate that modifying the payload, but using the same URN/labels
+    // does not change the shortId assignment.
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_DISTRIBUTION_DOUBLE);
+    testCases.add(KV.of("metric2", builder.build()));
+
+    builder = new SimpleMonitoringInfoBuilder(false);
+    builder.setUrn(MonitoringInfoConstants.Urns.USER_SUM_INT64);
+    builder.setLabel(MonitoringInfoConstants.Labels.NAME, "metricNumber7");
+    builder.setLabel(MonitoringInfoConstants.Labels.NAMESPACE, "myNamespace");
+    builder.setLabel(MonitoringInfoConstants.Labels.PTRANSFORM, "myPtransform");
+    testCases.add(KV.of("metric6", builder.build()));
+
+    // Verify each short ID is assigned properly.
+    Set<String> expectedShortIds = new HashSet<String>();
+    for (KV<String, MetricsApi.MonitoringInfo> entry : testCases) {
+      assertEquals(entry.getKey(), shortIdMap.getOrCreateShortId(entry.getValue()));
+      expectedShortIds.add(entry.getKey());
+    }
+
+    HashMap<String, MetricsApi.MonitoringInfo> actualRecoveredInfos = new HashMap<>();
+    for (String expectedShortId : expectedShortIds) {
+      actualRecoveredInfos.put(expectedShortId, shortIdMap.get(expectedShortId));
+    }
+    // Retrieve all of the MonitoringInfos by short id, and verify that the
+    // metadata (everything but the payload) matches the originals
+    assertEquals(expectedShortIds, actualRecoveredInfos.keySet());
+    for (KV<String, MetricsApi.MonitoringInfo> entry : testCases) {
+      // Clear payloads of both expected and actual before comparing
+      MetricsApi.MonitoringInfo expectedMonitoringInfo = entry.getValue();
+      MetricsApi.MonitoringInfo.Builder expected =
+          MetricsApi.MonitoringInfo.newBuilder(expectedMonitoringInfo);
+      expected.clearPayload();
+
+      MetricsApi.MonitoringInfo.Builder actual =
+          MetricsApi.MonitoringInfo.newBuilder(actualRecoveredInfos.get(entry.getKey()));
+      actual.clearPayload();
+      assertEquals(expected.build(), actual.build());
+    }
+
+    // Verify each short ID is assigned properly, in reverse.
+    for (int i = testCases.size() - 1; i > 0; i--) {
+      assertEquals(
+          testCases.get(i).getKey(), shortIdMap.getOrCreateShortId(testCases.get(i).getValue()));
+    }
+  }
+}

runners/flink/src/test/java/org/apache/beam/runners/flink/FlinkSavepointTest.java

@@ -68,7 +68,6 @@
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
-import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@@ -142,7 +141,6 @@ public void afterTest() throws Exception {
     ensureNoJobRunning();
   }
 
-  @Ignore("https://issues.apache.org/jira/projects/BEAM/issues/BEAM-10955")
   @Test
   public void testSavepointRestoreLegacy() throws Exception {
     runSavepointAndRestore(false);
@@ -193,7 +191,7 @@ private void runSavepointAndRestore(boolean isPortablePipeline) throws Exception
   private JobID executeLegacy(Pipeline pipeline) throws Exception {
     JobGraph jobGraph = getJobGraph(pipeline);
     flinkCluster.submitJob(jobGraph).get();
-    return jobGraph.getJobID();
+    return waitForJobToBeReady();
   }
 
   private JobID executePortable(Pipeline pipeline) throws Exception {

sdks/go/README.md

@@ -132,21 +132,28 @@ in a subdirectory of your GOPATH. This permits existing gradle tools to use your
 $ mkdir -p $GOPATH/src/github.com/apache/
 $ cd $GOPATH/src/github.com/apache/
 
-
 # Clone the repo, and update your branch as normal
 $ git clone https://github.com/apache/beam.git
 $ cd beam
 $ git remote add <GitHub_user> [email protected]:<GitHub_user>/beam.git
 $ git fetch --all
 
+# Navigate to the Go SDK root
+$ cd sdks/go
+# Set GO111MODULE to auto to allow building within a $GOPATH
+$ go env -w GO111MODULE=auto
 # Get or Update all the Go SDK dependencies
 $ go get -u ./...
-# Test that the system compiles and runs.
+# Test that the system compiles and runs
 $ go test ./...
 ```
 
 If you don’t have a GOPATH set, follow [these instructions](https://github.com/golang/go/wiki/SettingGOPATH) to create a new directory in your home directory, and use that.
 
+Developing the Go SDK doesn't support Go versions 1.17 or greater at this time.
+Once [BEAM-5379](https://issues.apache.org/jira/browse/BEAM-5379) is resolved,
+and the SDK properly supports Go Modules this should be simpler.
+
 Follow the [contribution guide](https://beam.apache.org/contribute/contribution-guide/#code) to create branches, and submit pull requests as normal.
 
 ### Dependency management

sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go

@@ -81,7 +81,7 @@ const (
 	// be found in Java's KafkaIO documentation.
 	LogAppendTime policy = "LogAppendTime"
 
-	readURN  = "beam:external:java:kafka:read:v1"
+	readURN  = "beam:external:java:kafkaio:typedwithoutmetadata:v1"
 	writeURN = "beam:external:java:kafka:write:v1"
 )
 

sdks/go/pkg/beam/testing/passert/equals.go

@@ -40,6 +40,19 @@ func Equals(s beam.Scope, col beam.PCollection, values ...interface{}) beam.PCol
 	return equals(subScope, col, other)
 }
 
+// EqualsList verifies that the given collection has the same values as a
+// given list, under coder equality. The values must be provided as an
+// array or slice. This is equivalent to passing a beam.CreateList PCollection
+// to Equals.
+func EqualsList(s beam.Scope, col beam.PCollection, list interface{}) beam.PCollection {
+	subScope := s.Scope("passert.EqualsList")
+	if list == nil {
+		return Empty(subScope, col)
+	}
+	listCollection := beam.CreateList(subScope, list)
+	return equals(subScope, col, listCollection)
+}
+
 // equals verifies that the actual values match the expected ones.
 func equals(s beam.Scope, actual, expected beam.PCollection) beam.PCollection {
 	unexpected, correct, missing := Diff(s, actual, expected)
@@ -100,3 +113,4 @@ func readToStrings(iter func(*beam.T) bool) []string {
 	sort.Strings(out)
 	return out
 }
+