Skip to content

Revert "Overhaul of SegmentPullers to add consistency and retries" #1250

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fjy merged 1 commit into from
Mar 30, 2015
Merged

Revert "Overhaul of SegmentPullers to add consistency and retries" #1250

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 1 addition & 11 deletions common/src/main/java/com/fasterxml/jackson/databind/introspect/GuiceInjectableValues.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop

This file was deleted.

36 changes: 5 additions & 31 deletions extensions/hdfs-storage/pom.xml
View file
Open in desktop
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