From c4ddedfb55bd3ca063bf7336b474e126ac262ea2 Mon Sep 17 00:00:00 2001 From: Davor Bonaci Date: Wed, 20 Apr 2016 16:02:46 -0700 Subject: [PATCH 1/2] Fix how this SDK identifies itself to the Cloud Dataflow service * Use Beam SDK naming, instead of Dataflow SDK. * Fix version lookup, so it correctly identifies the version. --- .../java/org/apache/beam/sdk/util/DataflowReleaseInfo.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/DataflowReleaseInfo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/DataflowReleaseInfo.java index 8c096fb7695c..e7d98b79de02 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/DataflowReleaseInfo.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/DataflowReleaseInfo.java @@ -34,7 +34,7 @@ public final class DataflowReleaseInfo extends GenericJson { private static final Logger LOG = LoggerFactory.getLogger(DataflowReleaseInfo.class); private static final String DATAFLOW_PROPERTIES_PATH = - "/org.apache.beam/sdk/sdk.properties"; + "/org/apache/beam/sdk/sdk.properties"; private static class LazyInit { private static final DataflowReleaseInfo INSTANCE = @@ -48,7 +48,7 @@ public static DataflowReleaseInfo getReleaseInfo() { return LazyInit.INSTANCE; } - @Key private String name = "Google Cloud Dataflow Java SDK"; + @Key private String name = "Apache Beam SDK for Java"; @Key private String version = "Unknown"; /** Provides the SDK name. */ From b130e80f8a37123e841e23c1fd060d953e78cd3c Mon Sep 17 00:00:00 2001 From: Davor Bonaci Date: Thu, 21 Apr 2016 10:08:53 -0700 Subject: [PATCH 2/2] Rename DataflowReleaseInfo to ReleaseInfo --- .../sdk/runners/DataflowPipelineRunner.java | 6 +++--- .../runners/DataflowPipelineRunnerTest.java | 6 +++--- .../beam/sdk/io/bigtable/BigtableIO.java | 4 ++-- ...aflowReleaseInfo.java => ReleaseInfo.java} | 20 +++++++++---------- 4 files changed, 18 insertions(+), 18 deletions(-) rename sdks/java/core/src/main/java/org/apache/beam/sdk/util/{DataflowReleaseInfo.java => ReleaseInfo.java} (77%) diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/sdk/runners/DataflowPipelineRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/sdk/runners/DataflowPipelineRunner.java index 55fc74815e3c..f88899c9f95d 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/sdk/runners/DataflowPipelineRunner.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/sdk/runners/DataflowPipelineRunner.java @@ -89,7 +89,6 @@ import org.apache.beam.sdk.transforms.windowing.GlobalWindows; import org.apache.beam.sdk.transforms.windowing.Window; import org.apache.beam.sdk.util.CoderUtils; -import org.apache.beam.sdk.util.DataflowReleaseInfo; import org.apache.beam.sdk.util.DataflowTransport; import org.apache.beam.sdk.util.IOChannelUtils; import org.apache.beam.sdk.util.InstanceBuilder; @@ -97,6 +96,7 @@ import org.apache.beam.sdk.util.PCollectionViews; import org.apache.beam.sdk.util.PathValidator; import org.apache.beam.sdk.util.PropertyNames; +import org.apache.beam.sdk.util.ReleaseInfo; import org.apache.beam.sdk.util.Reshuffle; import org.apache.beam.sdk.util.SystemDoFnInternal; import org.apache.beam.sdk.util.ValueWithRecordId; @@ -507,10 +507,10 @@ public DataflowPipelineJob run(Pipeline pipeline) { Job newJob = jobSpecification.getJob(); newJob.setClientRequestId(requestId); - String version = DataflowReleaseInfo.getReleaseInfo().getVersion(); + String version = ReleaseInfo.getReleaseInfo().getVersion(); System.out.println("Dataflow SDK version: " + version); - newJob.getEnvironment().setUserAgent(DataflowReleaseInfo.getReleaseInfo()); + newJob.getEnvironment().setUserAgent(ReleaseInfo.getReleaseInfo()); // The Dataflow Service may write to the temporary directory directly, so // must be verified. if (!Strings.isNullOrEmpty(options.getTempLocation())) { diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/sdk/runners/DataflowPipelineRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/sdk/runners/DataflowPipelineRunnerTest.java index 8b024fb8726c..79f38a024a28 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/sdk/runners/DataflowPipelineRunnerTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/sdk/runners/DataflowPipelineRunnerTest.java @@ -68,9 +68,9 @@ import org.apache.beam.sdk.transforms.windowing.IntervalWindow; import org.apache.beam.sdk.transforms.windowing.PaneInfo; import org.apache.beam.sdk.util.CoderUtils; -import org.apache.beam.sdk.util.DataflowReleaseInfo; import org.apache.beam.sdk.util.GcsUtil; import org.apache.beam.sdk.util.NoopPathValidator; +import org.apache.beam.sdk.util.ReleaseInfo; import org.apache.beam.sdk.util.TestCredential; import org.apache.beam.sdk.util.UserCodeException; import org.apache.beam.sdk.util.WindowedValue; @@ -375,10 +375,10 @@ public void testRunWithFiles() throws IOException { cloudDataflowDataset, workflowJob.getEnvironment().getDataset()); assertEquals( - DataflowReleaseInfo.getReleaseInfo().getName(), + ReleaseInfo.getReleaseInfo().getName(), workflowJob.getEnvironment().getUserAgent().get("name")); assertEquals( - DataflowReleaseInfo.getReleaseInfo().getVersion(), + ReleaseInfo.getReleaseInfo().getVersion(), workflowJob.getEnvironment().getUserAgent().get("version")); } diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/bigtable/BigtableIO.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/bigtable/BigtableIO.java index b2d9cb34fb54..5177262b57b8 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/bigtable/BigtableIO.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/bigtable/BigtableIO.java @@ -35,7 +35,7 @@ import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.runners.PipelineRunner; import org.apache.beam.sdk.transforms.PTransform; -import org.apache.beam.sdk.util.DataflowReleaseInfo; +import org.apache.beam.sdk.util.ReleaseInfo; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PBegin; import org.apache.beam.sdk.values.PCollection; @@ -978,7 +978,7 @@ public BigtableWriteException(KV> record, Throwab */ private static String getUserAgent() { String javaVersion = System.getProperty("java.specification.version"); - DataflowReleaseInfo info = DataflowReleaseInfo.getReleaseInfo(); + ReleaseInfo info = ReleaseInfo.getReleaseInfo(); return String.format( "%s/%s (%s); %s", info.getName(), diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/DataflowReleaseInfo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ReleaseInfo.java similarity index 77% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/DataflowReleaseInfo.java rename to sdks/java/core/src/main/java/org/apache/beam/sdk/util/ReleaseInfo.java index e7d98b79de02..77289ac63192 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/DataflowReleaseInfo.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/ReleaseInfo.java @@ -28,23 +28,23 @@ import java.util.Properties; /** - * Utilities for working with the Dataflow distribution. + * Utilities for working with release information. */ -public final class DataflowReleaseInfo extends GenericJson { - private static final Logger LOG = LoggerFactory.getLogger(DataflowReleaseInfo.class); +public final class ReleaseInfo extends GenericJson { + private static final Logger LOG = LoggerFactory.getLogger(ReleaseInfo.class); - private static final String DATAFLOW_PROPERTIES_PATH = + private static final String PROPERTIES_PATH = "/org/apache/beam/sdk/sdk.properties"; private static class LazyInit { - private static final DataflowReleaseInfo INSTANCE = - new DataflowReleaseInfo(DATAFLOW_PROPERTIES_PATH); + private static final ReleaseInfo INSTANCE = + new ReleaseInfo(PROPERTIES_PATH); } /** * Returns an instance of DataflowReleaseInfo. */ - public static DataflowReleaseInfo getReleaseInfo() { + public static ReleaseInfo getReleaseInfo() { return LazyInit.INSTANCE; } @@ -61,11 +61,11 @@ public String getVersion() { return version; } - private DataflowReleaseInfo(String resourcePath) { + private ReleaseInfo(String resourcePath) { Properties properties = new Properties(); - InputStream in = DataflowReleaseInfo.class.getResourceAsStream( - DATAFLOW_PROPERTIES_PATH); + InputStream in = ReleaseInfo.class.getResourceAsStream( + PROPERTIES_PATH); if (in == null) { LOG.warn("Dataflow properties resource not found: {}", resourcePath); return;