[GitHub] [beam] pabloem commented on a change in pull request #12083: [BEAM-10317] Java - Update BigQueryIO to tag BigQuery Jobs with the Dataflow Job ID

[GitBox <gi...@apache.org>]

pabloem commented on a change in pull request #12083:
URL: https://github.com/apache/beam/pull/12083#discussion_r448140534



##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -216,11 +221,11 @@ public void startQueryJob(JobReference jobRef, JobConfigurationQuery queryConfig
     @Override
     public void startCopyJob(JobReference jobRef, JobConfigurationTableCopy copyConfig)
         throws IOException, InterruptedException {
-      Job job =
-          new Job()
-              .setJobReference(jobRef)
-              .setConfiguration(new JobConfiguration().setCopy(copyConfig));
-
+      Map<String, String> labelMap = new HashMap<>();
+      JobConfiguration config = new JobConfiguration();
+      config.setCopy(copyConfig);
+      config.setLabels(this.bqIOMetadata.addAdditionalJobLabels(labelMap));
+      Job job = new Job().setJobReference(jobRef).setConfiguration(config);

Review comment:
       Same for other occurrences in this file

##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -216,11 +221,11 @@ public void startQueryJob(JobReference jobRef, JobConfigurationQuery queryConfig
     @Override
     public void startCopyJob(JobReference jobRef, JobConfigurationTableCopy copyConfig)
         throws IOException, InterruptedException {
-      Job job =
-          new Job()
-              .setJobReference(jobRef)
-              .setConfiguration(new JobConfiguration().setCopy(copyConfig));
-
+      Map<String, String> labelMap = new HashMap<>();
+      JobConfiguration config = new JobConfiguration();
+      config.setCopy(copyConfig);
+      config.setLabels(this.bqIOMetadata.addAdditionalJobLabels(labelMap));
+      Job job = new Job().setJobReference(jobRef).setConfiguration(config);

Review comment:
       I believe the protos might be immutable, and creating new instances when you call `setX` on them. You can do the whole thing fluently:
   
   ```suggestion
         Job job = new Job()
               .setJobReference(jobRef)
               .setConfiguration(new JobConfiguration()
                     .setCopy(copyConfig)
                     .setLabels(this.bqIOMetadata.addAdditionalJobLabels(labelMap)));
   ```

##########
File path: sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GceMetadataUtil.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.sdk.extensions.gcp.util;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.nio.charset.Charset;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.io.CharStreams;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpConnectionParams;
+import org.apache.http.params.HttpParams;
+
+/** */
+public class GceMetadataUtil {
+  private static final String BASE_METADATA_URL = "http://metadata/computeMetadata/v1/";
+
+  static String fetchMetadata(String key) {
+    int timeoutMillis = 5000;
+    final HttpParams httpParams = new BasicHttpParams();
+    HttpConnectionParams.setConnectionTimeout(httpParams, timeoutMillis);
+    HttpClient client = new DefaultHttpClient(httpParams);
+    HttpGet request = new HttpGet(BASE_METADATA_URL + key);
+    request.setHeader("Metadata-Flavor", "Google");
+
+    try {
+      HttpResponse response = client.execute(request);
+      if (response.getStatusLine().getStatusCode() != 200) {
+        // May mean its running on a non DataflowRunner, in which case it's perfectly normal.
+        return "";
+      }
+      InputStream in = response.getEntity().getContent();
+      try (final Reader reader = new InputStreamReader(in, Charset.defaultCharset())) {
+        return CharStreams.toString(reader);

Review comment:
       An intersting example: What if the worker running this code is a GCE VM, but not on Dataflow (e.g. Dataproc, DirectRunner on GCE VM)? Will that case be handled properly? This should be exercised by postcommits, I'd think...




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org