Skip to content

Commit

Permalink
Merge pull request apache#1250 from druid-io/revert-1132-uriSegmentLo…
Browse files Browse the repository at this point in the history 
…aders

Revert "Overhaul of SegmentPullers to add consistency and retries"
  • Loading branch information
@fjy
fjy committed Mar 30, 2015
2 parents 1e38def + f904bc7 commit cb3e9f7
Show file tree
Hide file tree
Showing 30 changed files with 276 additions and 1,800 deletions.
12 changes: 1 addition & 11 deletions common/src/main/java/com/fasterxml/jackson/databind/introspect/GuiceInjectableValues.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,6 @@
import com.fasterxml.jackson.databind.InjectableValues;
import com.google.inject.Injector;
import com.google.inject.Key;
import com.metamx.common.IAE;

import java.lang.reflect.Type;

/**
*/
Expand All @@ -39,13 +36,6 @@ public Object findInjectableValue(
Object valueId, DeserializationContext ctxt, BeanProperty forProperty, Object beanInstance
)
{
// From the docs: "Object that identifies value to inject; may be a simple name or more complex identifier object,
// whatever provider needs"
// Currently we should only be dealing with `Key` instances, and anything more advanced should be handled with
// great care
if(valueId instanceof Key){
return injector.getInstance((Key) valueId);
}
throw new IAE("Unknown class type [%s] for valueId [%s]", valueId.getClass().getCanonicalName(), valueId.toString());
return injector.getInstance((Key) valueId);
}
}
128 changes: 51 additions & 77 deletions ...assandra-storage/src/main/java/io/druid/storage/cassandra/CassandraDataSegmentPuller.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,105 +17,79 @@

package io.druid.storage.cassandra;

import com.google.common.base.Predicates;
import com.google.common.io.Files;
import com.google.inject.Inject;
import com.metamx.common.CompressionUtils;
import com.metamx.common.ISE;
import com.metamx.common.RetryUtils;
import com.metamx.common.logger.Logger;
import com.netflix.astyanax.recipes.storage.ChunkedStorage;
import com.netflix.astyanax.recipes.storage.ObjectMetadata;
import io.druid.segment.loading.DataSegmentPuller;
import io.druid.segment.loading.SegmentLoadingException;
import io.druid.timeline.DataSegment;
import io.druid.utils.CompressionUtils;
import org.apache.commons.io.FileUtils;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.Callable;

/**
* Cassandra Segment Puller
*
* @author boneill42
*/
public class CassandraDataSegmentPuller extends CassandraStorage implements DataSegmentPuller
{
private static final Logger log = new Logger(CassandraDataSegmentPuller.class);
private static final int CONCURRENCY = 10;
private static final int BATCH_SIZE = 10;
private static final Logger log = new Logger(CassandraDataSegmentPuller.class);
private static final int CONCURRENCY = 10;
private static final int BATCH_SIZE = 10;

@Inject
public CassandraDataSegmentPuller(CassandraDataSegmentConfig config)
{
super(config);
}
public CassandraDataSegmentPuller(CassandraDataSegmentConfig config)
{
super(config);
}

@Override
public void getSegmentFiles(DataSegment segment, File outDir) throws SegmentLoadingException
{
String key = (String) segment.getLoadSpec().get("key");
getSegmentFiles(key, outDir);
}
public com.metamx.common.FileUtils.FileCopyResult getSegmentFiles(final String key, final File outDir) throws SegmentLoadingException{
log.info("Pulling index from C* at path[%s] to outDir[%s]", key, outDir);
if (!outDir.exists()) {
outDir.mkdirs();
}
@Override
public void getSegmentFiles(DataSegment segment, File outDir) throws SegmentLoadingException
{
String key = (String) segment.getLoadSpec().get("key");
log.info("Pulling index from C* at path[%s] to outDir[%s]", key, outDir);

if (!outDir.isDirectory()) {
throw new ISE("outDir[%s] must be a directory.", outDir);
}
if (!outDir.exists())
{
outDir.mkdirs();
}

long startTime = System.currentTimeMillis();
final File tmpFile = new File(outDir, "index.zip");
log.info("Pulling to temporary local cache [%s]", tmpFile.getAbsolutePath());
if (!outDir.isDirectory())
{
throw new ISE("outDir[%s] must be a directory.", outDir);
}

final com.metamx.common.FileUtils.FileCopyResult localResult;
try {
localResult = RetryUtils.retry(
new Callable<com.metamx.common.FileUtils.FileCopyResult>()
{
@Override
public com.metamx.common.FileUtils.FileCopyResult call() throws Exception
{
try (OutputStream os = new FileOutputStream(tmpFile)) {
final ObjectMetadata meta = ChunkedStorage
.newReader(indexStorage, key, os)
.withBatchSize(BATCH_SIZE)
.withConcurrencyLevel(CONCURRENCY)
.call();
}
return new com.metamx.common.FileUtils.FileCopyResult(tmpFile);
}
},
Predicates.<Throwable>alwaysTrue(),
10
);
}catch (Exception e){
throw new SegmentLoadingException(e, "Unable to copy key [%s] to file [%s]", key, tmpFile.getAbsolutePath());
}
try{
final com.metamx.common.FileUtils.FileCopyResult result = CompressionUtils.unzip(tmpFile, outDir);
log.info(
"Pull of file[%s] completed in %,d millis (%s bytes)", key, System.currentTimeMillis() - startTime,
result.size()
);
return result;
}
catch (Exception e) {
try {
FileUtils.deleteDirectory(outDir);
}
catch (IOException e1) {
log.error(e1, "Error clearing segment directory [%s]", outDir.getAbsolutePath());
e.addSuppressed(e1);
}
throw new SegmentLoadingException(e, e.getMessage());
} finally {
if(!tmpFile.delete()){
log.warn("Could not delete cache file at [%s]", tmpFile.getAbsolutePath());
}
}
}
long startTime = System.currentTimeMillis();
ObjectMetadata meta = null;
final File outFile = new File(outDir, "index.zip");
try
{
try
{
log.info("Writing to [%s]", outFile.getAbsolutePath());
OutputStream os = Files.newOutputStreamSupplier(outFile).getOutput();
meta = ChunkedStorage
.newReader(indexStorage, key, os)
.withBatchSize(BATCH_SIZE)
.withConcurrencyLevel(CONCURRENCY)
.call();
os.close();
CompressionUtils.unzip(outFile, outDir);
} catch (Exception e)
{
FileUtils.deleteDirectory(outDir);
}
} catch (Exception e)
{
throw new SegmentLoadingException(e, e.getMessage());
}
log.info("Pull of file[%s] completed in %,d millis (%s bytes)", key, System.currentTimeMillis() - startTime,
meta.getObjectSize());
}
}
2 changes: 1 addition & 1 deletion ...assandra-storage/src/main/java/io/druid/storage/cassandra/CassandraDataSegmentPusher.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,14 +21,14 @@
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.inject.Inject;
import com.metamx.common.CompressionUtils;
import com.metamx.common.logger.Logger;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.recipes.storage.ChunkedStorage;
import io.druid.segment.SegmentUtils;
import io.druid.segment.loading.DataSegmentPusher;
import io.druid.segment.loading.DataSegmentPusherUtil;
import io.druid.timeline.DataSegment;
import io.druid.utils.CompressionUtils;

import java.io.File;
import java.io.FileInputStream;
Expand Down
43 changes: 10 additions & 33 deletions ...ions/cassandra-storage/src/main/java/io/druid/storage/cassandra/CassandraDruidModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package io.druid.storage.cassandra;

import com.fasterxml.jackson.core.Version;
import com.fasterxml.jackson.databind.Module;
import com.google.common.collect.ImmutableList;
import com.google.inject.Binder;
import com.google.inject.Key;
Expand All @@ -34,47 +34,24 @@
*/
public class CassandraDruidModule implements DruidModule
{
public static final String SCHEME = "c*";
@Override
public List<? extends Module> getJacksonModules()
{
return ImmutableList.of();
}

@Override
public void configure(Binder binder)
{
Binders.dataSegmentPullerBinder(binder)
.addBinding(SCHEME)
.to(CassandraDataSegmentPuller.class)
.in(LazySingleton.class);
.addBinding("c*")
.to(CassandraDataSegmentPuller.class)
.in(LazySingleton.class);

PolyBind.optionBinder(binder, Key.get(DataSegmentPusher.class))
.addBinding(SCHEME)
.addBinding("c*")
.to(CassandraDataSegmentPusher.class)
.in(LazySingleton.class);
JsonConfigProvider.bind(binder, "druid.storage", CassandraDataSegmentConfig.class);
}

@Override
public List<? extends com.fasterxml.jackson.databind.Module> getJacksonModules()
{
return ImmutableList.of(
new com.fasterxml.jackson.databind.Module()
{
@Override
public String getModuleName()
{
return "DruidCassandraStorage-" + System.identityHashCode(this);
}

@Override
public Version version()
{
return Version.unknownVersion();
}

@Override
public void setupModule(SetupContext context)
{
context.registerSubtypes(CassandraLoadSpec.class);
}
}
);
}
}
54 changes: 0 additions & 54 deletions extensions/cassandra-storage/src/main/java/io/druid/storage/cassandra/CassandraLoadSpec.java
View file

This file was deleted.

36 changes: 5 additions & 31 deletions extensions/hdfs-storage/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,39 +63,13 @@
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
</dependency>

<!-- Tests -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.druid</groupId>
<artifactId>druid-server</artifactId>
<version>${parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.3.0</version>
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.3.0</version>
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>2.3.0</version>
<scope>test</scope>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</dependencies>

</project>
Loading

0 comments on commit cb3e9f7

Please sign in to comment.