remove datasource from hadoop output path (#3196)

fixes #2083, follow-up to #1702

api/src/main/java/io/druid/segment/loading/DataSegmentPusher.java

@@ -26,6 +26,8 @@ import java.io.IOException;
 
 public interface DataSegmentPusher
 {
-  public String getPathForHadoop(String dataSource);
-  public DataSegment push(File file, DataSegment segment) throws IOException;
+  @Deprecated
+  String getPathForHadoop(String dataSource);
+  String getPathForHadoop();
+  DataSegment push(File file, DataSegment segment) throws IOException;
 }

extensions-contrib/azure-extensions/src/main/java/io/druid/storage/azure/AzureDataSegmentPusher.java

@@ -58,8 +58,15 @@ public class AzureDataSegmentPusher implements DataSegmentPusher
     this.jsonMapper = jsonMapper;
   }
 
+  @Deprecated
   @Override
   public String getPathForHadoop(String dataSource)
+  {
+    return getPathForHadoop();
+  }
+
+  @Override
+  public String getPathForHadoop()
   {
     return null;
   }

extensions-contrib/cassandra-storage/src/main/java/io/druid/storage/cassandra/CassandraDataSegmentPusher.java

@@ -58,17 +58,24 @@ public class CassandraDataSegmentPusher extends CassandraStorage implements Data
 	}
 
   @Override
-  public String getPathForHadoop(String dataSource)
+  public String getPathForHadoop()
   {
     throw new UnsupportedOperationException("Cassandra storage does not support indexing via Hadoop");
   }
 
+  @Deprecated
   @Override
-	public DataSegment push(final File indexFilesDir, DataSegment segment) throws IOException
-	{
-		log.info("Writing [%s] to C*", indexFilesDir);
-		String key = JOINER.join(
-		    config.getKeyspace().isEmpty() ? null : config.getKeyspace(),
+  public String getPathForHadoop(String dataSource)
+  {
+    return getPathForHadoop();
+  }
+
+  @Override
+  public DataSegment push(final File indexFilesDir, DataSegment segment) throws IOException
+  {
+    log.info("Writing [%s] to C*", indexFilesDir);
+    String key = JOINER.join(
+        config.getKeyspace().isEmpty() ? null : config.getKeyspace(),
 		    DataSegmentPusherUtil.getStorageDir(segment)
 		    );
 

extensions-contrib/cloudfiles-extensions/src/main/java/io/druid/storage/cloudfiles/CloudFilesDataSegmentPusher.java

@@ -59,11 +59,18 @@ public class CloudFilesDataSegmentPusher implements DataSegmentPusher
   }
 
   @Override
-  public String getPathForHadoop(final String dataSource)
+  public String getPathForHadoop()
   {
     return null;
   }
 
+  @Deprecated
+  @Override
+  public String getPathForHadoop(final String dataSource)
+  {
+    return getPathForHadoop();
+  }
+
   @Override
   public DataSegment push(final File indexFilesDir, final DataSegment inSegment) throws IOException
   {

extensions-core/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsDataSegmentPusher.java

@@ -63,8 +63,15 @@ public class HdfsDataSegmentPusher implements DataSegmentPusher
     log.info("Configured HDFS as deep storage");
   }
 
+  @Deprecated
   @Override
   public String getPathForHadoop(String dataSource)
+  {
+    return getPathForHadoop();
+  }
+
+  @Override
+  public String getPathForHadoop()
   {
     return new Path(config.getStorageDirectory()).toUri().toString();
   }

extensions-core/s3-extensions/src/main/java/io/druid/storage/s3/S3DataSegmentPusher.java

@@ -61,10 +61,17 @@ public class S3DataSegmentPusher implements DataSegmentPusher
     log.info("Configured S3 as deep storage");
   }
 
+  @Override
+  public String getPathForHadoop()
+  {
+    return String.format("s3n://%s/%s", config.getBucket(), config.getBaseKey());
+  }
+
+  @Deprecated
   @Override
   public String getPathForHadoop(String dataSource)
   {
-    return String.format("s3n://%s/%s/%s", config.getBucket(), config.getBaseKey(), dataSource);
+    return getPathForHadoop();
   }
 
   @Override

indexing-service/src/main/java/io/druid/indexing/common/task/HadoopIndexTask.java

@@ -177,7 +177,7 @@ public class HadoopIndexTask extends HadoopTask
         new String[]{
             toolbox.getObjectMapper().writeValueAsString(spec),
             toolbox.getConfig().getHadoopWorkingPath(),
-            toolbox.getSegmentPusher().getPathForHadoop(getDataSource())
+            toolbox.getSegmentPusher().getPathForHadoop()
         },
         loader
     );

indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java

@@ -339,10 +339,17 @@ public class IndexTask extends AbstractFixedIntervalTask
     final List<DataSegment> pushedSegments = new CopyOnWriteArrayList<DataSegment>();
     final DataSegmentPusher wrappedDataSegmentPusher = new DataSegmentPusher()
     {
+      @Deprecated
       @Override
       public String getPathForHadoop(String dataSource)
       {
-        return toolbox.getSegmentPusher().getPathForHadoop(dataSource);
+        return getPathForHadoop();
+      }
+
+      @Override
+      public String getPathForHadoop()
+      {
+        return toolbox.getSegmentPusher().getPathForHadoop();
       }
 
       @Override

indexing-service/src/test/java/io/druid/indexing/common/task/IndexTaskTest.java

@@ -242,8 +242,15 @@ public class IndexTaskTest
           }
         }, null, new DataSegmentPusher()
         {
+          @Deprecated
           @Override
           public String getPathForHadoop(String dataSource)
+          {
+            return getPathForHadoop();
+          }
+
+          @Override
+          public String getPathForHadoop()
           {
             return null;
           }

indexing-service/src/test/java/io/druid/indexing/firehose/IngestSegmentFirehoseFactoryTest.java

@@ -213,8 +213,15 @@ public class IngestSegmentFirehoseFactoryTest
         newMockEmitter(),
         new DataSegmentPusher()
         {
+          @Deprecated
           @Override
           public String getPathForHadoop(String dataSource)
+          {
+            return getPathForHadoop();
+          }
+
+          @Override
+          public String getPathForHadoop()
           {
             throw new UnsupportedOperationException();
           }

indexing-service/src/test/java/io/druid/indexing/overlord/TaskLifecycleTest.java

@@ -456,11 +456,18 @@ public class TaskLifecycleTest
     return new DataSegmentPusher()
     {
       @Override
-      public String getPathForHadoop(String dataSource)
+      public String getPathForHadoop()
       {
         throw new UnsupportedOperationException();
       }
 
+      @Deprecated
+      @Override
+      public String getPathForHadoop(String dataSource)
+      {
+        return getPathForHadoop();
+      }
+
       @Override
       public DataSegment push(File file, DataSegment segment) throws IOException
       {
@@ -993,8 +1000,15 @@ public class TaskLifecycleTest
   {
     dataSegmentPusher = new DataSegmentPusher()
     {
+      @Deprecated
       @Override
       public String getPathForHadoop(String s)
+      {
+        return getPathForHadoop();
+      }
+
+      @Override
+      public String getPathForHadoop()
       {
         throw new UnsupportedOperationException();
       }

indexing-service/src/test/java/io/druid/indexing/test/TestDataSegmentPusher.java

@@ -32,8 +32,15 @@ public class TestDataSegmentPusher implements DataSegmentPusher
 {
   private final Set<DataSegment> pushedSegments = Sets.newConcurrentHashSet();
 
+  @Deprecated
   @Override
   public String getPathForHadoop(String dataSource)
+  {
+    return getPathForHadoop();
+  }
+
+  @Override
+  public String getPathForHadoop()
   {
     throw new UnsupportedOperationException();
   }

server/src/main/java/io/druid/segment/loading/LocalDataSegmentPusher.java

@@ -53,10 +53,17 @@ public class LocalDataSegmentPusher implements DataSegmentPusher
     log.info("Configured local filesystem as deep storage");
   }
 
+  @Override
+  public String getPathForHadoop()
+  {
+    return config.getStorageDirectory().getAbsoluteFile().toURI().toString();
+  }
+
+  @Deprecated
   @Override
   public String getPathForHadoop(String dataSource)
   {
-    return new File(config.getStorageDirectory().getAbsoluteFile(), dataSource).toURI().toString();
+    return getPathForHadoop();
   }
 
   @Override

server/src/test/java/io/druid/segment/loading/LocalDataSegmentPusherTest.java

@@ -120,8 +120,8 @@ public class LocalDataSegmentPusherTest
     config.storageDirectory = new File("/druid");
 
     Assert.assertEquals(
-        "file:/druid/foo",
-        new LocalDataSegmentPusher(config, new ObjectMapper()).getPathForHadoop("foo")
+        "file:/druid",
+        new LocalDataSegmentPusher(config, new ObjectMapper()).getPathForHadoop()
     );
   }
 
@@ -131,8 +131,8 @@ public class LocalDataSegmentPusherTest
     config.storageDirectory = new File("druid");
 
     Assert.assertEquals(
-        String.format("file:%s/druid/foo", System.getProperty("user.dir")),
-        new LocalDataSegmentPusher(config, new ObjectMapper()).getPathForHadoop("foo")
+        String.format("file:%s/druid", System.getProperty("user.dir")),
+        new LocalDataSegmentPusher(config, new ObjectMapper()).getPathForHadoop()
     );
   }
 }

server/src/test/java/io/druid/segment/realtime/appenderator/AppenderatorTester.java

@@ -165,8 +165,15 @@ public class AppenderatorTester implements AutoCloseable
     EmittingLogger.registerEmitter(emitter);
     dataSegmentPusher = new DataSegmentPusher()
     {
+      @Deprecated
       @Override
       public String getPathForHadoop(String dataSource)
+      {
+        return getPathForHadoop();
+      }
+
+      @Override
+      public String getPathForHadoop()
       {
         throw new UnsupportedOperationException();
       }

services/src/main/java/io/druid/cli/CliRealtimeExample.java

@@ -121,10 +121,18 @@ public class CliRealtimeExample extends ServerRunnable
 
   private static class NoopDataSegmentPusher implements DataSegmentPusher
   {
+
+    @Override
+    public String getPathForHadoop()
+    {
+      return "noop";
+    }
+
+    @Deprecated
     @Override
     public String getPathForHadoop(String dataSource)
     {
-      return dataSource;
+      return getPathForHadoop();
     }
 
     @Override