[virgil] r154 committed - Exception mapper for keyspaces that already exist....
virgil.apach...@codespot.com
Author: boneill42
Date: Tue Jan 24 13:31:50 2012
Log: Exception mapper for keyspaces that already exist.
Work on AOP for triggers.
http://code.google.com/a/apache-extras.org/p/virgil/source/detail?r=154
Added:
/trunk/server/src/main/java/org/apache/virgil/aop
/trunk/server/src/main/java/org/apache/virgil/aop/CassandraDaemonAspect.aj
/trunk/server/src/main/java/org/apache/virgil/aop/CassandraServerAspect.aj
/trunk/server/src/main/java/org/apache/virgil/exception
/trunk/server/src/main/java/org/apache/virgil/exception/KeyspaceExceptionMapper.java
/trunk/server/src/main/java/org/apache/virgil/triggers
/trunk/server/src/main/java/org/apache/virgil/triggers/DistributedCommitLog.java
Modified:
/trunk/mapreduce/src/main/java/org/apache/virgil/mapreduce/RubyMapReduce.java
/trunk/release/assembly/bin/virgil-hadoop
/trunk/release/assembly/mapreduce/jars/virgil-mapreduce-hdeploy.jar
/trunk/server/pom.xml
/trunk/server/src/main/java/org/apache/virgil/CassandraStorage.java
/trunk/server/src/main/java/org/apache/virgil/VirgilService.java
/trunk/server/src/main/java/org/apache/virgil/cli/VirgilCommand.java
/trunk/server/src/main/java/org/apache/virgil/config/VirgilConfiguration.java
/trunk/server/src/main/java/org/apache/virgil/health/CassandraHealthCheck.java
/trunk/server/src/main/java/org/apache/virgil/index/SolrIndexer.java
/trunk/server/src/main/java/org/apache/virgil/resource/DataResource.java
/trunk/server/src/main/java/org/apache/virgil/resource/MapReduceResource.java
/trunk/server/src/test/resources/virgil.yaml
=======================================
--- /dev/null
+++
/trunk/server/src/main/java/org/apache/virgil/aop/CassandraDaemonAspect.aj
Tue Jan 24 13:31:50 2012
@@ -0,0 +1,16 @@
+package org.apache.virgil.aop;
+
+public aspect CassandraDaemonAspect {
+
+ private pointcut mainMethod() :
+ execution(public static void main(String[]));
+
+ before() : mainMethod() {
+ //throw new RuntimeException("CRASH IT!");
+ System.out.println("> " + thisJoinPoint);
+ }
+
+ after() : mainMethod() {
+ System.out.println("< " + thisJoinPoint);
+ }
+}
=======================================
--- /dev/null
+++
/trunk/server/src/main/java/org/apache/virgil/aop/CassandraServerAspect.aj
Tue Jan 24 13:31:50 2012
@@ -0,0 +1,50 @@
+package org.apache.virgil.aop;
+
+import java.util.List;
+
+import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.IColumn;
+import org.apache.cassandra.db.IMutation;
+import org.apache.cassandra.db.RowMutation;
+import org.apache.cassandra.thrift.ConsistencyLevel;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.virgil.triggers.DistributedCommitLog;
+
+public aspect CassandraServerAspect {
+
+ private pointcut insertMethod() :
+ execution(private void doInsert(ConsistencyLevel, List<?
extends IMutation>));
+
+ after() : insertMethod() {
+ try {
+ Object consistencyLevel = thisJoinPoint.getArgs()[0];
+ @SuppressWarnings("unchecked")
+ List<IMutation> mutations = (List<IMutation>)
thisJoinPoint.getArgs()[1];
+ for (IMutation mutation : mutations) {
+ if (mutation instanceof RowMutation) {
+ RowMutation rowMutation = (RowMutation) mutation;
+ String key = ByteBufferUtil.string(rowMutation.key());
+ System.out.println("Mutation for [" + key + "] @ [" +
rowMutation.getTable()
+ + "] with consistencyLevel [" +
consistencyLevel + "]");
+ for (Integer cfId : rowMutation.getColumnFamilyIds()) {
+ ColumnFamily columnFamily =
rowMutation.getColumnFamily(cfId);
+ for (IColumn column :
columnFamily.getSortedColumns()) {
+ String name =
ByteBufferUtil.string(column.name());
+ String value =
ByteBufferUtil.string(column.value());
+ boolean delete =
columnFamily.isMarkedForDelete();
+ if (delete) {
+ System.out.println(" -- DELETE -- [" +
name + "] => [" + value + "]");
+ } else {
+ System.out.println(" -- SET -- [" + name
+ "] => [" + value + "]");
+ }
+ }
+ }
+ }
+ DistributedCommitLog.writeMutation(mutation);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+
+ }
+}
=======================================
--- /dev/null
+++
/trunk/server/src/main/java/org/apache/virgil/exception/KeyspaceExceptionMapper.java
Tue Jan 24 13:31:50 2012
@@ -0,0 +1,18 @@
+//
+// Copyright (c) 2012 Health Market Science, Inc.
+//
+package org.apache.virgil.exception;
+
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import javax.ws.rs.ext.ExceptionMapper;
+import javax.ws.rs.ext.Provider;
+
+import org.apache.cassandra.thrift.InvalidRequestException;
+
+@Provider
+public class KeyspaceExceptionMapper implements
ExceptionMapper<InvalidRequestException> {
+ public Response toResponse(InvalidRequestException exception) {
+ return
Response.status(Status.OK).entity(exception.getWhy()).build();
+ }
+}
=======================================
--- /dev/null
+++
/trunk/server/src/main/java/org/apache/virgil/triggers/DistributedCommitLog.java
Tue Jan 24 13:31:50 2012
@@ -0,0 +1,48 @@
+package org.apache.virgil.triggers;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.cassandra.db.IMutation;
+import org.apache.cassandra.thrift.CfDef;
+import org.apache.cassandra.thrift.KsDef;
+import org.apache.virgil.CassandraStorage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DistributedCommitLog {
+ public static final String KEYSPACE = "cirrus";
+ public static final String COLUMN_FAMILY = "CommitLog";
+ private static boolean initialized = false;
+ private static Logger logger =
LoggerFactory.getLogger(DistributedCommitLog.class);
+
+ public static void create() {
+ if (!initialized) {
+ try {
+ List<CfDef> cfDefList = new ArrayList<CfDef>();
+ KsDef ksDef = new
KsDef(KEYSPACE, "org.apache.cassandra.locator.SimpleStrategy", cfDefList);
+ ksDef.putToStrategy_options("replication_factor", "1");
+
CassandraStorage.getCassandra(null).system_add_keyspace(ksDef);
+ } catch (Exception e) {
+ logger.warn("Did not create System.CommitLog. (probably
already there)");
+ }
+
+ try {
+ CfDef columnFamily = new CfDef(KEYSPACE, COLUMN_FAMILY);
+ columnFamily.setKey_validation_class("UTF8Type");
+ columnFamily.setComparator_type("UTF8Type");
+ columnFamily.setDefault_validation_class("UTF8Type");
+
CassandraStorage.getCassandra(KEYSPACE).system_add_column_family(columnFamily);
+ initialized = true;
+ } catch (Exception e) {
+ e.printStackTrace();
+ logger.warn("Did not create System.CommitLog. (probably
already there)");
+ }
+
+ }
+ }
+
+ public static void writeMutation(IMutation mutation) {
+ DistributedCommitLog.create();
+ }
+}
=======================================
---
/trunk/mapreduce/src/main/java/org/apache/virgil/mapreduce/RubyMapReduce.java
Tue Jan 17 12:30:34 2012
+++
/trunk/mapreduce/src/main/java/org/apache/virgil/mapreduce/RubyMapReduce.java
Tue Jan 24 13:31:50 2012
@@ -170,6 +170,7 @@
protected void setup(Context context) throws IOException,
InterruptedException {
String source = context.getConfiguration().get("source");
this.rubyContainer = new
ScriptingContainer(LocalContextScope.CONCURRENT);
+ this.rubyContainer.setLoadPaths(getGemPaths());
this.rubyReceiver = rubyContainer.runScriptlet(source);
if (context.getConfiguration().get("params") != null) {
params = new HashMap<String, Object>();
=======================================
--- /trunk/release/assembly/bin/virgil-hadoop Mon Jan 16 18:38:55 2012
+++ /trunk/release/assembly/bin/virgil-hadoop Tue Jan 24 13:31:50 2012
@@ -35,6 +35,7 @@
JAVA=`which java`
fi
-CLASSPATH=$CLASSPATH:$cwd/../conf/;$cwd/../mapreduce/conf/
+CLASSPATH=$CLASSPATH:$cwd/../conf/
+CLASSPATH=$CLASSPATH:$cwd/../mapreduce/conf/
$JAVA -cp $CLASSPATH org.apache.virgil.VirgilService server
conf/virgil.yaml $@
=======================================
--- /trunk/release/assembly/mapreduce/jars/virgil-mapreduce-hdeploy.jar Thu
Jan 19 12:07:55 2012
+++ /trunk/release/assembly/mapreduce/jars/virgil-mapreduce-hdeploy.jar Tue
Jan 24 13:31:50 2012
File is too large to display a diff.
=======================================
--- /trunk/server/pom.xml Fri Jan 20 07:08:32 2012
+++ /trunk/server/pom.xml Tue Jan 24 13:31:50 2012
@@ -31,6 +31,12 @@
<artifactId>commons-httpclient</artifactId>
<version>3.1</version>
</dependency>
+
+ <dependency>
+ <groupId>org.aspectj</groupId>
+ <artifactId>aspectjrt</artifactId>
+ <version>1.6.11</version>
+ </dependency>
</dependencies>
<build>
@@ -40,6 +46,33 @@
<directory>${basedir}/src/main/resources</directory>
</resource>
</resources>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>aspectj-maven-plugin</artifactId>
+ <version>1.4</version>
+ <configuration>
+
<complianceLevel>1.5</complianceLevel>
+ <weaveDependencies>
+ <weaveDependency>
+ <groupId>org.apache.cassandra</groupId>
+ <artifactId>cassandra-thrift</artifactId>
+ </weaveDependency>
+ <weaveDependency>
+ <groupId>org.apache.cassandra</groupId>
+ <artifactId>cassandra-all</artifactId>
+ </weaveDependency>
+ </weaveDependencies>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
</build>
</project>
=======================================
--- /trunk/server/src/main/java/org/apache/virgil/CassandraStorage.java Mon
Jan 16 18:38:55 2012
+++ /trunk/server/src/main/java/org/apache/virgil/CassandraStorage.java Tue
Jan 24 13:31:50 2012
@@ -50,46 +50,29 @@
import org.json.simple.JSONObject;
public class CassandraStorage {
- private static final int MAX_COLUMNS = 1000;
+ private static final int MAX_COLUMNS = 1000;
private static final int MAX_ROWS = 20;
- // private static Cassandra.Iface server = null;
private Indexer indexer = null;
private VirgilConfiguration config = null;
- private String host = null;
- private int port;
- private boolean embedded = false;
// TODO: Come back and make indexing AOP
- public CassandraStorage(String host, int port, VirgilConfiguration
config, Indexer indexer, boolean embedded) {
+ public CassandraStorage(VirgilConfiguration config, Indexer indexer) {
this.indexer = indexer;
// CassandraStorage.server = server;
this.config = config;
- this.host = host;
- this.port = port;
- this.embedded = embedded;
- }
-
- public String getHost() {
- return host;
- }
-
- public int getPort() {
- return port;
- }
-
- public boolean isEmbedded(){
- return this.embedded;
}
// For now, get a new connection every time.
- private Cassandra.Iface getCassandra(String keyspace) throws Exception
{
- if (embedded) {
+ // TODO: Use connection pooling
+ public static Cassandra.Iface getCassandra(String keyspace)
+ throws Exception {
+ if (VirgilConfiguration.isEmbedded()) {
Cassandra.Iface server = new CassandraServer();
if (keyspace != null)
server.set_keyspace(keyspace);
return server;
} else {
- TTransport tr = new TFramedTransport(new TSocket(this.host,
this.port));
+ TTransport tr = new TFramedTransport(new
TSocket(VirgilConfiguration.getHost(), VirgilConfiguration.getPort()));
TProtocol proto = new TBinaryProtocol(tr);
tr.open();
Cassandra.Iface server = new Cassandra.Client(proto);
=======================================
--- /trunk/server/src/main/java/org/apache/virgil/VirgilService.java Mon
Jan 16 18:38:55 2012
+++ /trunk/server/src/main/java/org/apache/virgil/VirgilService.java Tue
Jan 24 13:31:50 2012
@@ -2,6 +2,7 @@
import org.apache.virgil.cli.VirgilCommand;
import org.apache.virgil.config.VirgilConfiguration;
+import org.apache.virgil.exception.KeyspaceExceptionMapper;
import org.apache.virgil.health.CassandraHealthCheck;
import org.apache.virgil.resource.DataResource;
import org.apache.virgil.resource.MapReduceResource;
@@ -29,6 +30,7 @@
env.addResource(new MapReduceResource(this));
env.addResource(new DataResource(this));
env.addHealthCheck(new CassandraHealthCheck(this));
+ env.addProvider(new KeyspaceExceptionMapper());
}
public CassandraStorage getStorage() {
=======================================
--- /trunk/server/src/main/java/org/apache/virgil/cli/VirgilCommand.java
Mon Jan 16 18:38:55 2012
+++ /trunk/server/src/main/java/org/apache/virgil/cli/VirgilCommand.java
Tue Jan 24 13:31:50 2012
@@ -47,20 +47,22 @@
System.setProperty("cassandra-foreground", "true");
System.setProperty(VirgilConfiguration.CASSANDRA_PORT_PROPERTY, "9160");
System.setProperty(VirgilConfiguration.CASSANDRA_HOST_PROPERTY, "localhost");
+
System.setProperty(VirgilConfiguration.CASSANDRA_EMBEDDED, "1");
CassandraDaemon.main(null);
- return new CassandraStorage("localhost", 9160, config, indexer, true);
+ return new CassandraStorage(config, indexer);
} else {
String cassandraHost = params.getOptionValue("host");
if (cassandraHost == null)
throw new RuntimeException("Need to specify a host if not running in
embedded mode. (-e)");
- System.setProperty(VirgilConfiguration.CASSANDRA_HOST_PROPERTY,
cassandraHost);
String cassandraPort = params.getOptionValue("port");
if (cassandraPort == null)
cassandraPort = "9160";
+
System.setProperty(VirgilConfiguration.CASSANDRA_HOST_PROPERTY,
cassandraHost);
System.setProperty(VirgilConfiguration.CASSANDRA_PORT_PROPERTY,
cassandraPort);
+
System.setProperty(VirgilConfiguration.CASSANDRA_EMBEDDED, "0");
System.out.println("Starting virgil against remote cassandra server ["
+ cassandraHost + ":"
+ cassandraPort + "]");
- return new CassandraStorage(cassandraHost,
Integer.parseInt(cassandraPort), config, indexer, false);
+ return new CassandraStorage(config, indexer);
}
}
=======================================
---
/trunk/server/src/main/java/org/apache/virgil/config/VirgilConfiguration.java
Mon Jan 16 18:14:34 2012
+++
/trunk/server/src/main/java/org/apache/virgil/config/VirgilConfiguration.java
Tue Jan 24 13:31:50 2012
@@ -10,6 +10,7 @@
public class VirgilConfiguration extends Configuration {
public final static String CASSANDRA_HOST_PROPERTY
= "virgil.cassandra_host";
public final static String CASSANDRA_PORT_PROPERTY
= "virgil.cassandra_port";
+ public final static String CASSANDRA_EMBEDDED = "virgil.embedded";
@NotEmpty
@NotNull
@@ -40,4 +41,16 @@
else
return ConsistencyLevel.valueOf(consistencyLevel);
}
-}
+
+ public static boolean isEmbedded(){
+ return
System.getProperty(VirgilConfiguration.CASSANDRA_EMBEDDED).equals("1");
+ }
+
+ public static String getHost(){
+ return
System.getProperty(VirgilConfiguration.CASSANDRA_HOST_PROPERTY);
+ }
+
+ public static int getPort(){
+ return
Integer.parseInt(System.getProperty(VirgilConfiguration.CASSANDRA_PORT_PROPERTY));
+ }
+}
=======================================
---
/trunk/server/src/main/java/org/apache/virgil/health/CassandraHealthCheck.java
Mon Jan 16 18:38:55 2012
+++
/trunk/server/src/main/java/org/apache/virgil/health/CassandraHealthCheck.java
Tue Jan 24 13:31:50 2012
@@ -1,7 +1,9 @@
package org.apache.virgil.health;
import org.apache.virgil.VirgilService;
+import org.apache.virgil.config.VirgilConfiguration;
import org.json.simple.JSONArray;
+
import com.yammer.metrics.core.HealthCheck;
public class CassandraHealthCheck extends HealthCheck {
@@ -17,8 +19,8 @@
Result result = null;
try {
- String host = this.service.getStorage().getHost();
- int port = this.service.getStorage().getPort();
+ String host = VirgilConfiguration.getHost();
+ int port = VirgilConfiguration.getPort();
JSONArray keyspaces = this.service.getStorage().getKeyspaces();
String output = "Connected to [" + host + ":" + port + "] w/ " +
keyspaces.size() + " keyspaces.";
result = Result.healthy(output);
=======================================
--- /trunk/server/src/main/java/org/apache/virgil/index/SolrIndexer.java
Mon Jan 16 18:14:34 2012
+++ /trunk/server/src/main/java/org/apache/virgil/index/SolrIndexer.java
Tue Jan 24 13:31:50 2012
@@ -9,65 +9,62 @@
import org.json.simple.JSONValue;
public class SolrIndexer implements Indexer {
- public static String SOLR_CONFIG_PARAM = "solr_host";
- public static String CONTENT_TYPE = "application/json";
- public static String XML_CONTENT_TYPE = "application/xml";
- public static String CHAR_SET = "UTF8";
- // TODO: Maybe move dynamic field suffix to config file?
- public static String DYNAMIC_FIELD_SUFFIX = "_t";
- private String solrUrl = null;
-
- public SolrIndexer(VirgilConfiguration config) {
- solrUrl = config.getSolrHost();
- }
-
- @Override
- public void index(String columnFamily, String rowKey, String json) throws
Exception {
- JSONObject row = (JSONObject) JSONValue.parse(json);
- index(columnFamily, rowKey, row);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public void index(String columnFamily, String rowKey, JSONObject row)
throws Exception {
- HttpClient client = new HttpClient();
- PostMethod post = new PostMethod(solrUrl + "update/json?commit=true");
- JSONObject document = new JSONObject();
-
- document.put("id", this.getDocumentId(columnFamily, rowKey));
- document.put("rowKey" + DYNAMIC_FIELD_SUFFIX, rowKey);
- document.put("columnFamily" + DYNAMIC_FIELD_SUFFIX, columnFamily);
- for (Object column : row.keySet()) {
- document.put(column.toString().toLowerCase() + DYNAMIC_FIELD_SUFFIX,
row.get(column));
- }
-
- // Index
- RequestEntity requestEntity = new StringRequestEntity("[" +
document.toString() + "]", CONTENT_TYPE, CHAR_SET);
- post.setRequestEntity(requestEntity);
- try {
- client.executeMethod(post);
- } finally {
- post.releaseConnection();
- }
- }
-
- @Override
- public void delete(String columnFamily, String rowKey) throws Exception {
- HttpClient client = new HttpClient();
-
- // Commit
- PostMethod post = new PostMethod(solrUrl + "update?commit=true");
- String query = "id:" + this.getDocumentId(columnFamily, rowKey);
- StringRequestEntity requestEntity = new
StringRequestEntity("<delete><query>" + query + "</query></delete>",
- XML_CONTENT_TYPE, CHAR_SET);
- post.setRequestEntity(requestEntity);
- client.executeMethod(post);
- post.releaseConnection();
- }
-
- // TODO: Could hash the rowkey and column then combine to avoid potential
- // collisions.
- private String getDocumentId(String columnFamily, String rowKey) {
- return columnFamily + "." + rowKey;
- }
-}
+ public static String SOLR_CONFIG_PARAM = "solr_host";
+ public static String CONTENT_TYPE = "application/json";
+ public static String XML_CONTENT_TYPE = "application/xml";
+ public static String CHAR_SET = "UTF8";
+ // TODO: Maybe move dynamic field suffix to config file?
+ public static String DYNAMIC_FIELD_SUFFIX = "_t";
+ private String solrUrl = null;
+
+ public SolrIndexer(VirgilConfiguration config) {
+ solrUrl = config.getSolrHost();
+ }
+
+ public void index(String columnFamily, String rowKey, String json)
throws Exception {
+ JSONObject row = (JSONObject) JSONValue.parse(json);
+ index(columnFamily, rowKey, row);
+ }
+
+ @SuppressWarnings("unchecked")
+ public void index(String columnFamily, String rowKey, JSONObject row)
throws Exception {
+ HttpClient client = new HttpClient();
+ PostMethod post = new PostMethod(solrUrl
+ "update/json?commit=true");
+ JSONObject document = new JSONObject();
+
+ document.put("id", this.getDocumentId(columnFamily, rowKey));
+ document.put("rowKey" + DYNAMIC_FIELD_SUFFIX, rowKey);
+ document.put("columnFamily" + DYNAMIC_FIELD_SUFFIX, columnFamily);
+ for (Object column : row.keySet()) {
+ document.put(column.toString().toLowerCase() +
DYNAMIC_FIELD_SUFFIX, row.get(column));
+ }
+
+ // Index
+ RequestEntity requestEntity = new StringRequestEntity("[" +
document.toString() + "]", CONTENT_TYPE, CHAR_SET);
+ post.setRequestEntity(requestEntity);
+ try {
+ client.executeMethod(post);
+ } finally {
+ post.releaseConnection();
+ }
+ }
+
+ public void delete(String columnFamily, String rowKey) throws
Exception {
+ HttpClient client = new HttpClient();
+
+ // Commit
+ PostMethod post = new PostMethod(solrUrl + "update?commit=true");
+ String query = "id:" + this.getDocumentId(columnFamily, rowKey);
+ StringRequestEntity requestEntity = new
StringRequestEntity("<delete><query>" + query + "</query></delete>",
+ XML_CONTENT_TYPE, CHAR_SET);
+ post.setRequestEntity(requestEntity);
+ client.executeMethod(post);
+ post.releaseConnection();
+ }
+
+ // TODO: Could hash the rowkey and column then combine to avoid
potential
+ // collisions.
+ private String getDocumentId(String columnFamily, String rowKey) {
+ return columnFamily + "." + rowKey;
+ }
+}
=======================================
---
/trunk/server/src/main/java/org/apache/virgil/resource/DataResource.java
Fri Jan 20 10:30:28 2012
+++
/trunk/server/src/main/java/org/apache/virgil/resource/DataResource.java
Tue Jan 24 13:31:50 2012
@@ -9,7 +9,6 @@
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
-import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.virgil.CassandraStorage;
import org.apache.virgil.VirgilService;
import org.apache.virgil.config.VirgilConfiguration;
@@ -31,19 +30,6 @@
this.virgilService = virgilService;
this.config = virgilService.getConfig();
}
-
- //
================================================================================================================
- // Key Space Operations
- //
================================================================================================================
- @GET
- @Path("/ping")
- @Produces({ "text/plain" })
- public String healthCheck() throws Exception {
- if (logger.isDebugEnabled())
- logger.debug("Ping.");
- getCassandraStorage().getSlice("system", "Versions", "build",
ConsistencyLevel.ONE);
- return "OK\n";
- }
//
================================================================================================================
// Key Space Operations
=======================================
---
/trunk/server/src/main/java/org/apache/virgil/resource/MapReduceResource.java
Mon Jan 16 20:39:41 2012
+++
/trunk/server/src/main/java/org/apache/virgil/resource/MapReduceResource.java
Tue Jan 24 13:31:50 2012
@@ -7,6 +7,7 @@
import org.apache.virgil.CassandraStorage;
import org.apache.virgil.VirgilService;
+import org.apache.virgil.config.VirgilConfiguration;
import org.apache.virgil.mapreduce.JobSpawner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,13 +48,13 @@
logger.debug(" <-- Output : Keyspace [" + outputKeyspace
+ "], ColumnFamily [" + outputColumnFamily + "]");
}
- if (this.virgilService.getStorage().isEmbedded()) {
+ if (VirgilConfiguration.isEmbedded()) {
logger.debug("Running in embedded mode.");
- JobSpawner.spawnLocal(jobName,
this.getCassandraStorage().getHost(), this.getCassandraStorage().getPort(),
- inputKeyspace, inputColumnFamily, outputKeyspace,
outputColumnFamily, source, params);
+ JobSpawner.spawnLocal(jobName, VirgilConfiguration.getHost(),
VirgilConfiguration.getPort(), inputKeyspace,
+ inputColumnFamily, outputKeyspace, outputColumnFamily,
source, params);
} else {
logger.debug("Spawning job remotely.");
- JobSpawner.spawnRemote(jobName,
this.getCassandraStorage().getHost(), this.getCassandraStorage().getPort(),
+ JobSpawner.spawnRemote(jobName, VirgilConfiguration.getHost(),
VirgilConfiguration.getPort(),
inputKeyspace, inputColumnFamily, outputKeyspace,
outputColumnFamily, source, params);
}
}
=======================================
--- /trunk/server/src/test/resources/virgil.yaml Mon Jan 16 18:14:34 2012
+++ /trunk/server/src/test/resources/virgil.yaml Tue Jan 24 13:31:50 2012
@@ -121,9 +121,10 @@
loggers:
# Sets the level for 'com.example.app' to DEBUG.
- org.apache.virgil: DEBUG
+ org.apache.virgil: WARN
org.apache: WARN
org.eclipse: WARN
+ httpclient.wire: WARN
# Settings for logging to stdout.
console: