Delagate creation of segmentPath/LoadSpec to DataSegmentPushers and add S3a support

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

@@ -19,15 +19,45 @@
 
 package io.druid.segment.loading;
 
+import com.google.common.base.Joiner;
 import io.druid.timeline.DataSegment;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
 
 public interface DataSegmentPusher
 {
+  Joiner JOINER = Joiner.on("/").skipNulls();
+
   @Deprecated
   String getPathForHadoop(String dataSource);
   String getPathForHadoop();
   DataSegment push(File file, DataSegment segment) throws IOException;
+  //use map instead of LoadSpec class to avoid dependency pollution.
+  Map<String, Object> makeLoadSpec(URI finalIndexZipFilePath);
+  default String getStorageDir(DataSegment dataSegment) {
+    return getDefaultStorageDir(dataSegment);
+  }
+  default String makeIndexPathName(DataSegment dataSegment, String indexName) {
+    return String.format("./%s/%s", getStorageDir(dataSegment),indexName);
+  }
+
+  // Note: storage directory structure format = .../dataSource/interval/version/partitionNumber/
+  // If above format is ever changed, make sure to change it appropriately in other places
+  // e.g. HDFSDataSegmentKiller uses this information to clean the version, interval and dataSource directories
+  // on segment deletion if segment being deleted was the only segment
+  static String getDefaultStorageDir(DataSegment segment) {
+    return JOINER.join(
+        segment.getDataSource(),
+        String.format(
+            "%s_%s",
+            segment.getInterval().getStart(),
+            segment.getInterval().getEnd()
+        ),
+        segment.getVersion(),
+        segment.getShardSpec().getPartitionNum()
+    );
+  }
 }

examples/conf-quickstart/druid/middleManager/runtime.properties

@@ -17,4 +17,4 @@ druid.processing.numThreads=2
 
 # Hadoop indexing
 druid.indexer.task.hadoopWorkingPath=var/druid/hadoop-tmp
-druid.indexer.task.defaultHadoopCoordinates=["org.apache.hadoop:hadoop-client:2.3.0"]
+druid.indexer.task.defaultHadoopCoordinates=["org.apache.hadoop:hadoop-client:2.7.3"]

examples/conf/druid/middleManager/runtime.properties

@@ -17,4 +17,4 @@ druid.processing.numThreads=2
 
 # Hadoop indexing
 druid.indexer.task.hadoopWorkingPath=var/druid/hadoop-tmp
-druid.indexer.task.defaultHadoopCoordinates=["org.apache.hadoop:hadoop-client:2.3.0"]
+druid.indexer.task.defaultHadoopCoordinates=["org.apache.hadoop:hadoop-client:2.7.3"]

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

@@ -29,12 +29,12 @@
 import io.druid.java.util.common.logger.Logger;
 import io.druid.segment.SegmentUtils;
 import io.druid.segment.loading.DataSegmentPusher;
-import io.druid.segment.loading.DataSegmentPusherUtil;
 import io.druid.timeline.DataSegment;
 
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Map;
 import java.util.concurrent.Callable;
@@ -85,7 +85,7 @@ public File createSegmentDescriptorFile(final ObjectMapper jsonMapper, final Dat
 
   public Map<String, String> getAzurePaths(final DataSegment segment)
   {
-    final String storageDir = DataSegmentPusherUtil.getStorageDir(segment);
+    final String storageDir = this.getStorageDir(segment);
 
     return ImmutableMap.of(
         "index", String.format("%s/%s", storageDir, AzureStorageDruidModule.INDEX_ZIP_FILE_NAME),
@@ -109,16 +109,7 @@ public DataSegment uploadDataSegment(
 
     final DataSegment outSegment = segment
         .withSize(size)
-        .withLoadSpec(
-            ImmutableMap.<String, Object>of(
-                "type",
-                AzureStorageDruidModule.SCHEME,
-                "containerName",
-                config.getContainer(),
-                "blobPath",
-                azurePaths.get("index")
-            )
-        )
+        .withLoadSpec(this.makeLoadSpec(new URI(azurePaths.get("index"))))
         .withBinaryVersion(version);
 
     log.info("Deleting file [%s]", compressedSegmentData);
@@ -174,4 +165,17 @@ public DataSegment call() throws Exception
       }
     }
   }
+
+  @Override
+  public Map<String, Object> makeLoadSpec(URI uri)
+  {
+    return ImmutableMap.<String, Object>of(
+        "type",
+        AzureStorageDruidModule.SCHEME,
+        "containerName",
+        config.getContainer(),
+        "blobPath",
+        uri.toString()
+    );
+  }
 }

extensions-contrib/azure-extensions/src/test/java/io/druid/storage/azure/AzureDataSegmentPusherTest.java

@@ -27,7 +27,6 @@
 import com.microsoft.azure.storage.StorageException;
 import io.druid.jackson.DefaultObjectMapper;
 import io.druid.java.util.common.MapUtils;
-import io.druid.segment.loading.DataSegmentPusherUtil;
 import io.druid.timeline.DataSegment;
 import io.druid.timeline.partition.NoneShardSpec;
 import org.easymock.EasyMockSupport;
@@ -112,9 +111,9 @@ public void testPush() throws Exception
   @Test
   public void getAzurePathsTest()
   {
-    final String storageDir = DataSegmentPusherUtil.getStorageDir(dataSegment);
-    AzureDataSegmentPusher pusher = new AzureDataSegmentPusher(azureStorage, azureAccountConfig, jsonMapper);
 
+    AzureDataSegmentPusher pusher = new AzureDataSegmentPusher(azureStorage, azureAccountConfig, jsonMapper);
+    final String storageDir = pusher.getStorageDir(dataSegment);
     Map<String, String> paths = pusher.getAzurePaths(dataSegment);
 
     assertEquals(String.format("%s/%s", storageDir, AzureStorageDruidModule.INDEX_ZIP_FILE_NAME), paths.get("index"));

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

@@ -30,12 +30,13 @@
 import io.druid.java.util.common.logger.Logger;
 import io.druid.segment.SegmentUtils;
 import io.druid.segment.loading.DataSegmentPusher;
-import io.druid.segment.loading.DataSegmentPusherUtil;
 import io.druid.timeline.DataSegment;
 
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
 
 /**
  * Cassandra Segment Pusher
@@ -77,7 +78,7 @@ public DataSegment push(final File indexFilesDir, DataSegment segment) throws IO
     log.info("Writing [%s] to C*", indexFilesDir);
     String key = JOINER.join(
         config.getKeyspace().isEmpty() ? null : config.getKeyspace(),
-        DataSegmentPusherUtil.getStorageDir(segment)
+        this.getStorageDir(segment)
         );
 
     // Create index
@@ -114,4 +115,10 @@ ImmutableMap.<String, Object> of("type", "c*", "key", key)
     compressedIndexFile.delete();
     return segment;
   }
+
+  @Override
+  public Map<String, Object> makeLoadSpec(URI uri)
+  {
+    throw new UnsupportedOperationException("not supported");
+  }
 }

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

@@ -34,6 +34,8 @@
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
 import java.util.concurrent.Callable;
 
 public class CloudFilesDataSegmentPusher implements DataSegmentPusher
@@ -75,7 +77,7 @@ public String getPathForHadoop(final String dataSource)
   @Override
   public DataSegment push(final File indexFilesDir, final DataSegment inSegment) throws IOException
   {
-    final String segmentPath = CloudFilesUtils.buildCloudFilesPath(this.config.getBasePath(), inSegment);
+    final String segmentPath = CloudFilesUtils.buildCloudFilesPath(this.config.getBasePath(), getStorageDir(inSegment));
 
     File descriptorFile = null;
     File zipOutFile = null;
@@ -112,18 +114,7 @@ public DataSegment call() throws Exception
 
               final DataSegment outSegment = inSegment
                   .withSize(indexSize)
-                  .withLoadSpec(
-                      ImmutableMap.<String, Object>of(
-                          "type",
-                          CloudFilesStorageDruidModule.SCHEME,
-                          "region",
-                          segmentData.getRegion(),
-                          "container",
-                          segmentData.getContainer(),
-                          "path",
-                          segmentData.getPath()
-                      )
-                  )
+                  .withLoadSpec(makeLoadSpec(new URI(segmentData.getPath())))
                   .withBinaryVersion(SegmentUtils.getVersionFromDir(indexFilesDir));
 
               return outSegment;
@@ -146,4 +137,19 @@ public DataSegment call() throws Exception
       }
     }
   }
+
+  @Override
+  public Map<String, Object> makeLoadSpec(URI uri)
+  {
+    return ImmutableMap.<String, Object>of(
+        "type",
+        CloudFilesStorageDruidModule.SCHEME,
+        "region",
+        objectApi.getRegion(),
+        "container",
+        objectApi.getContainer(),
+        "path",
+        uri.toString()
+    );
+  }
 }

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

@@ -22,8 +22,6 @@
 import com.google.common.base.Predicate;
 
 import io.druid.java.util.common.RetryUtils;
-import io.druid.segment.loading.DataSegmentPusherUtil;
-import io.druid.timeline.DataSegment;
 
 import java.io.IOException;
 import java.util.concurrent.Callable;
@@ -70,9 +68,4 @@ public static String buildCloudFilesPath(String basePath, final String fileName)
     return path;
   }
 
-  public static String buildCloudFilesPath(String basePath, final DataSegment segment)
-  {
-    return buildCloudFilesPath(basePath, DataSegmentPusherUtil.getStorageDir(segment));
-  }
-
 }