src/main/java/org/embulk/input/gcs/GcsFileInputPlugin.java in embulk-input-gcs-0.1.13 vs src/main/java/org/embulk/input/gcs/GcsFileInputPlugin.java in embulk-input-gcs-0.2.0

@@ -50,11 +50,12 @@
     {
         @Config("bucket")
         String getBucket();
 
         @Config("path_prefix")
-        String getPathPrefix();
+        @ConfigDefault("null")
+        Optional<String> getPathPrefix();
 
         @Config("last_path")
         @ConfigDefault("null")
         Optional<String> getLastPath();
 
@@ -82,10 +83,12 @@
 
         @Config("json_keyfile")
         @ConfigDefault("null")
         Optional<LocalFile> getJsonKeyfile();
 
+        @Config("paths")
+        @ConfigDefault("[]")
         List<String> getFiles();
         void setFiles(List<String> files);
 
         @ConfigInject
         BufferAllocator getBufferAllocator();
@@ -129,12 +132,17 @@
             }
         }
 
         Storage client = newGcsClient(task, newGcsAuth(task));
 
-        // list files recursively
-        task.setFiles(listFiles(task, client));
+        // list files recursively if path_prefix is specified
+        if (task.getPathPrefix().isPresent()) {
+            task.setFiles(listFiles(task, client));
+        }
+        if (task.getFiles().isEmpty()) {
+            throw new ConfigException("No file is found. Fix path_prefix or specify paths directly");
+        }
         // number of processors is same with number of files
         return resume(task.dump(), task.getFiles().size(), control);
     }
 
     private GcsAuthentication newGcsAuth(PluginTask task)
@@ -212,10 +220,10 @@
 
     public List<String> listFiles(PluginTask task, Storage client)
     {
         String bucket = task.getBucket();
 
-        return listGcsFilesByPrefix(client, bucket, task.getPathPrefix(), task.getLastPath());
+        return listGcsFilesByPrefix(client, bucket, task.getPathPrefix().get(), task.getLastPath());
     }
 
     /**
      * Lists GCS filenames filtered by prefix.
      *