From 529cbd741246e7eaa62db5abd169076553efcd92 Mon Sep 17 00:00:00 2001
From: Eugene Kirpichov <kirpichov@google.com>
Date: Wed, 19 Apr 2017 16:10:45 -0700
Subject: [PATCH] Validates that input and output GCS paths specify a bucket

---
 .../apache/beam/sdk/util/GcsPathValidator.java |  3 +++
 .../beam/sdk/util/GcsPathValidatorTest.java    | 18 ++++++++++++++++++
 2 files changed, 21 insertions(+)

diff --git a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java
index a5b951de1aaac..f2e7a8372e19c 100644
--- a/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java
+++ b/sdks/java/extensions/gcp-core/src/main/java/org/apache/beam/sdk/util/GcsPathValidator.java
@@ -67,6 +67,9 @@ public String validateOutputFilePrefixSupported(String filePrefix) {
   public String verifyPath(String path) {
     GcsPath gcsPath = getGcsPath(path);
     checkArgument(gcsPath.isAbsolute(), "Must provide absolute paths for Dataflow");
+    checkArgument(!gcsPath.getObject().isEmpty(),
+        "Missing object or bucket in path: '%s', did you mean: 'gs://some-bucket/%s'?",
+        gcsPath, gcsPath.getBucket());
     checkArgument(!gcsPath.getObject().contains("//"),
         "Dataflow Service does not allow objects with consecutive slashes");
     return gcsPath.toResourceName();
diff --git a/sdks/java/extensions/gcp-core/src/test/java/org/apache/beam/sdk/util/GcsPathValidatorTest.java b/sdks/java/extensions/gcp-core/src/test/java/org/apache/beam/sdk/util/GcsPathValidatorTest.java
index dc363199c061d..d4c804aef834a 100644
--- a/sdks/java/extensions/gcp-core/src/test/java/org/apache/beam/sdk/util/GcsPathValidatorTest.java
+++ b/sdks/java/extensions/gcp-core/src/test/java/org/apache/beam/sdk/util/GcsPathValidatorTest.java
@@ -63,6 +63,15 @@ public void testInvalidFilePattern() {
     validator.validateInputFilePatternSupported("/local/path");
   }
 
+  @Test
+  public void testFilePatternMissingBucket() {
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage(
+        "Missing object or bucket in path: 'gs://input/', "
+            + "did you mean: 'gs://some-bucket/input'?");
+    validator.validateInputFilePatternSupported("gs://input");
+  }
+
   @Test
   public void testWhenBucketDoesNotExist() throws Exception {
     when(mockGcsUtil.bucketAccessible(any(GcsPath.class))).thenReturn(false);
@@ -84,4 +93,13 @@ public void testInvalidOutputPrefix() {
         "Expected a valid 'gs://' path but was given '/local/path'");
     validator.validateOutputFilePrefixSupported("/local/path");
   }
+
+  @Test
+  public void testOutputPrefixMissingBucket() {
+    expectedException.expect(IllegalArgumentException.class);
+    expectedException.expectMessage(
+        "Missing object or bucket in path: 'gs://output/', "
+            + "did you mean: 'gs://some-bucket/output'?");
+    validator.validateOutputFilePrefixSupported("gs://output");
+  }
 }