You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2012/02/07 21:59:36 UTC
svn commit: r1241622 - in /incubator/accumulo/trunk: ./ src/core/
src/core/src/main/java/org/apache/accumulo/core/zookeeper/
src/examples/wikisearch/ src/examples/wikisearch/ingest/bin/
src/examples/wikisearch/ingest/conf/ src/examples/wikisearch/inges...
Author: kturner
Date: Tue Feb 7 20:59:36 2012
New Revision: 1241622
URL: http://svn.apache.org/viewvc?rev=1241622&view=rev
Log:
ACCUMULO-366 ACCUMULO-379 ACCUMULO-375 merged from 1.4
Added:
incubator/accumulo/trunk/src/examples/wikisearch/README.parallel
- copied unchanged from r1241620, incubator/accumulo/branches/1.4/src/examples/wikisearch/README.parallel
incubator/accumulo/trunk/src/examples/wikisearch/ingest/bin/ingest_parallel.sh
- copied unchanged from r1241620, incubator/accumulo/branches/1.4/src/examples/wikisearch/ingest/bin/ingest_parallel.sh
incubator/accumulo/trunk/src/examples/wikisearch/ingest/conf/wikipedia_parallel.xml.example
- copied unchanged from r1241620, incubator/accumulo/branches/1.4/src/examples/wikisearch/ingest/conf/wikipedia_parallel.xml.example
incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedIngester.java
- copied unchanged from r1241620, incubator/accumulo/branches/1.4/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedIngester.java
incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedMapper.java
- copied unchanged from r1241620, incubator/accumulo/branches/1.4/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitionedMapper.java
incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitioner.java
- copied unchanged from r1241620, incubator/accumulo/branches/1.4/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaPartitioner.java
incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/test/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaInputSplitTest.java
- copied unchanged from r1241620, incubator/accumulo/branches/1.4/src/examples/wikisearch/ingest/src/test/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaInputSplitTest.java
Modified:
incubator/accumulo/trunk/ (props changed)
incubator/accumulo/trunk/src/core/ (props changed)
incubator/accumulo/trunk/src/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java
incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java
incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaInputFormat.java
incubator/accumulo/trunk/src/server/ (props changed)
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Module.java
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Node.java
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/bulk/Verify.java
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/BatchScan.java
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/ScanTable.java
incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/shard/VerifyIndex.java
Propchange: incubator/accumulo/trunk/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Feb 7 20:59:36 2012
@@ -1,3 +1,3 @@
/incubator/accumulo/branches/1.3:1190280,1190413,1190420,1190427,1190500,1195622,1195625,1195629,1195635,1196044,1196054,1196057,1196071-1196072,1196106,1197066,1198935,1199383,1203683,1204625,1205547,1205880,1206169,1208031,1209124,1209526,1209532,1209539,1209541,1209587,1209657,1210518,1210571,1210596,1210598,1213424,1214320,1225006,1227215,1227231,1227611,1228195,1230180,1230736,1231043,1236873
/incubator/accumulo/branches/1.3.5rc:1209938
-/incubator/accumulo/branches/1.4:1201902-1241519
+/incubator/accumulo/branches/1.4:1201902-1241620
Propchange: incubator/accumulo/trunk/src/core/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Feb 7 20:59:36 2012
@@ -1,3 +1,3 @@
/incubator/accumulo/branches/1.3.5rc/src/core:1209938
/incubator/accumulo/branches/1.3/src/core:1190280,1190413,1190420,1190427,1190500,1195622,1195625,1195629,1195635,1196044,1196054,1196057,1196071-1196072,1196106,1197066,1198935,1199383,1203683,1204625,1205547,1205880,1206169,1208031,1209124,1209526,1209532,1209539,1209541,1209587,1209657,1210518,1210571,1210596,1210598,1213424,1214320,1225006,1227215
-/incubator/accumulo/branches/1.4/src/core:1201902-1241519
+/incubator/accumulo/branches/1.4/src/core:1201902-1241620
Modified: incubator/accumulo/trunk/src/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java (original)
+++ incubator/accumulo/trunk/src/core/src/main/java/org/apache/accumulo/core/zookeeper/ZooCache.java Tue Feb 7 20:59:36 2012
@@ -136,7 +136,7 @@ public class ZooCache {
}
log.warn("Zookeeper error, will retry", e);
} catch (InterruptedException e) {
- log.warn("Zookeeper error, will retry", e);
+ log.info("Zookeeper error, will retry", e);
} catch (ConcurrentModificationException e) {
log.debug("Zookeeper was modified, will retry");
}
Modified: incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java (original)
+++ incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/ArticleExtractor.java Tue Feb 7 20:59:36 2012
@@ -16,6 +16,9 @@
*/
package org.apache.accumulo.examples.wikisearch.ingest;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
import java.io.Reader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
@@ -29,6 +32,7 @@ import javax.xml.stream.XMLStreamReader;
import org.apache.accumulo.examples.wikisearch.normalizer.LcNoDiacriticsNormalizer;
import org.apache.accumulo.examples.wikisearch.normalizer.NumberNormalizer;
+import org.apache.hadoop.io.Writable;
public class ArticleExtractor {
@@ -37,13 +41,15 @@ public class ArticleExtractor {
private static NumberNormalizer nn = new NumberNormalizer();
private static LcNoDiacriticsNormalizer lcdn = new LcNoDiacriticsNormalizer();
- public static class Article {
+ public static class Article implements Writable {
int id;
String title;
long timestamp;
String comments;
String text;
+ public Article(){}
+
private Article(int id, String title, long timestamp, String comments, String text) {
super();
this.id = id;
@@ -90,6 +96,24 @@ public class ArticleExtractor {
fields.put("COMMENTS", lcdn.normalizeFieldValue("COMMENTS", this.comments));
return fields;
}
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ id = in.readInt();
+ title = in.readUTF();
+ timestamp = in.readLong();
+ comments = in.readUTF();
+ text = in.readUTF();
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeInt(id);
+ out.writeUTF(title);
+ out.writeLong(timestamp);
+ out.writeUTF(comments);
+ out.writeUTF(text);
+ }
}
Modified: incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java (original)
+++ incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaConfiguration.java Tue Feb 7 20:59:36 2012
@@ -48,6 +48,11 @@ public class WikipediaConfiguration {
public final static String NUM_GROUPS = "wikipedia.ingest.groups";
+ public final static String PARTITIONED_ARTICLES_DIRECTORY = "wikipedia.partitioned.directory";
+
+ public final static String RUN_PARTITIONER = "wikipedia.run.partitioner";
+ public final static String RUN_INGEST = "wikipedia.run.ingest";
+
public static String getUser(Configuration conf) {
return conf.get(USER);
@@ -117,6 +122,18 @@ public class WikipediaConfiguration {
return conf.getInt(NUM_GROUPS, 1);
}
+ public static Path getPartitionedArticlesPath(Configuration conf) {
+ return new Path(conf.get(PARTITIONED_ARTICLES_DIRECTORY));
+ }
+
+ public static boolean runPartitioner(Configuration conf) {
+ return conf.getBoolean(RUN_PARTITIONER, false);
+ }
+
+ public static boolean runIngest(Configuration conf) {
+ return conf.getBoolean(RUN_INGEST, true);
+ }
+
/**
* Helper method to get properties from Hadoop configuration
*
Modified: incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaInputFormat.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaInputFormat.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaInputFormat.java (original)
+++ incubator/accumulo/trunk/src/examples/wikisearch/ingest/src/main/java/org/apache/accumulo/examples/wikisearch/ingest/WikipediaInputFormat.java Tue Feb 7 20:59:36 2012
@@ -75,10 +75,14 @@ public class WikipediaInputFormat extend
Path file = new Path(in.readUTF());
long start = in.readLong();
long length = in.readLong();
- int numHosts = in.readInt();
- String[] hosts = new String[numHosts];
- for(int i = 0; i < numHosts; i++)
- hosts[i] = in.readUTF();
+ String [] hosts = null;
+ if(in.readBoolean())
+ {
+ int numHosts = in.readInt();
+ hosts = new String[numHosts];
+ for(int i = 0; i < numHosts; i++)
+ hosts[i] = in.readUTF();
+ }
fileSplit = new FileSplit(file, start, length, hosts);
partition = in.readInt();
}
@@ -89,10 +93,17 @@ public class WikipediaInputFormat extend
out.writeLong(fileSplit.getStart());
out.writeLong(fileSplit.getLength());
String [] hosts = fileSplit.getLocations();
- out.writeInt(hosts.length);
- for(String host:hosts)
+ if(hosts == null)
+ {
+ out.writeBoolean(false);
+ }
+ else
+ {
+ out.writeBoolean(true);
+ out.writeInt(hosts.length);
+ for(String host:hosts)
out.writeUTF(host);
- fileSplit.write(out);
+ }
out.writeInt(partition);
}
Propchange: incubator/accumulo/trunk/src/server/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Feb 7 20:59:36 2012
@@ -1,3 +1,3 @@
/incubator/accumulo/branches/1.3.5rc/src/server:1209938
/incubator/accumulo/branches/1.3/src/server:1190280,1190413,1190420,1190427,1190500,1195622,1195625,1195629,1195635,1196044,1196054,1196057,1196071-1196072,1196106,1197066,1198935,1199383,1203683,1204625,1205547,1205880,1206169,1208031,1209124,1209526,1209532,1209539,1209541,1209587,1209657,1210518,1210571,1210596,1210598,1213424,1214320,1225006,1227215,1227231,1227611
-/incubator/accumulo/branches/1.4/src/server:1201902-1241519
+/incubator/accumulo/branches/1.4/src/server:1201902-1241620
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/Tablet.java Tue Feb 7 20:59:36 2012
@@ -123,6 +123,8 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
import cloudtrace.instrument.Span;
import cloudtrace.instrument.Trace;
@@ -2274,6 +2276,7 @@ public class Tablet {
if (updateMetadata) {
synchronized (this) {
updatingFlushID = false;
+ this.notifyAll();
}
}
}
@@ -2281,8 +2284,19 @@ public class Tablet {
}
boolean initiateMinorCompaction() {
+ if (isClosed()) {
+ // don't bother trying to get flush id if closed... could be closed after this check but that is ok... just trying to cut down on uneeded log messages....
+ return false;
+ }
+
// get the flush id before the new memmap is made available for write
- long flushId = getFlushID();
+ long flushId;
+ try {
+ flushId = getFlushID();
+ } catch (NoNodeException e) {
+ log.info("Asked to initiate MinC when there was no flush id " + getExtent() + " " + e.getMessage());
+ return false;
+ }
return initiateMinorCompaction(flushId);
}
@@ -2338,23 +2352,39 @@ public class Tablet {
return true;
}
- long getFlushID() {
+ long getFlushID() throws NoNodeException {
try {
String zTablePath = Constants.ZROOT + "/" + HdfsZooInstance.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ Constants.ZTABLE_FLUSH_ID;
return Long.parseLong(new String(ZooReaderWriter.getRetryingInstance().getData(zTablePath, null)));
- } catch (Exception e) {
+ } catch (InterruptedException e) {
throw new RuntimeException(e);
+ } catch (NumberFormatException nfe) {
+ throw new RuntimeException(nfe);
+ } catch (KeeperException ke) {
+ if (ke instanceof NoNodeException) {
+ throw (NoNodeException) ke;
+ } else {
+ throw new RuntimeException(ke);
+ }
}
}
- long getCompactionID() {
+ long getCompactionID() throws NoNodeException {
try {
String zTablePath = Constants.ZROOT + "/" + HdfsZooInstance.getInstance().getInstanceID() + Constants.ZTABLES + "/" + extent.getTableId()
+ Constants.ZTABLE_COMPACT_ID;
return Long.parseLong(new String(ZooReaderWriter.getRetryingInstance().getData(zTablePath, null)));
- } catch (Exception e) {
+ } catch (InterruptedException e) {
throw new RuntimeException(e);
+ } catch (NumberFormatException nfe) {
+ throw new RuntimeException(nfe);
+ } catch (KeeperException ke) {
+ if (ke instanceof NoNodeException) {
+ throw (NoNodeException) ke;
+ } else {
+ throw new RuntimeException(ke);
+ }
}
}
@@ -2557,13 +2587,25 @@ public class Tablet {
}
}
+ while (updatingFlushID) {
+ try {
+ this.wait(50);
+ } catch (InterruptedException e) {
+ log.error(e.toString());
+ }
+ }
+
if (!saveState || tabletMemory.getMemTable().getNumEntries() == 0) {
return;
}
tabletMemory.waitForMinC();
- mct = prepareForMinC(getFlushID());
+ try {
+ mct = prepareForMinC(getFlushID());
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
if (queueMinC) {
tabletResources.executeMinorCompaction(mct);
@@ -2612,7 +2654,11 @@ public class Tablet {
tabletMemory.waitForMinC();
if (saveState && tabletMemory.getMemTable().getNumEntries() > 0) {
- prepareForMinC(getFlushID()).run();
+ try {
+ prepareForMinC(getFlushID()).run();
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
}
if (saveState) {
@@ -3103,7 +3149,11 @@ public class Tablet {
Long compactionId = null;
if (!propogateDeletes) {
// compacting everything, so update the compaction id in !METADATA
- compactionId = getCompactionID();
+ try {
+ compactionId = getCompactionID();
+ } catch (NoNodeException e) {
+ throw new RuntimeException(e);
+ }
}
// need to handle case where only one file is being major compacted
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/tabletserver/TabletServer.java Tue Feb 7 20:59:36 2012
@@ -194,6 +194,7 @@ import org.apache.thrift.TProcessor;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.server.TServer;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NoNodeException;
import cloudtrace.instrument.Span;
import cloudtrace.instrument.Trace;
@@ -1887,7 +1888,13 @@ public class TabletServer extends Abstra
if (flushID == null) {
// read the flush id once from zookeeper instead of reading
// it for each tablet
- flushID = tablet.getFlushID();
+ try {
+ flushID = tablet.getFlushID();
+ } catch (NoNodeException e) {
+ // table was probably deleted
+ log.info("Asked to flush table that has no flush id " + ke + " " + e.getMessage());
+ return;
+ }
}
tablet.flush(flushID);
}
@@ -1904,7 +1911,11 @@ public class TabletServer extends Abstra
Tablet tablet = onlineTablets.get(new KeyExtent(textent));
if (tablet != null) {
log.info("Flushing " + tablet.getExtent());
- tablet.flush(tablet.getFlushID());
+ try {
+ tablet.flush(tablet.getFlushID());
+ } catch (NoNodeException nne) {
+ log.info("Asked to flush tablet that has no flush id " + new KeyExtent(textent) + " " + nne.getMessage());
+ }
}
}
@@ -1999,7 +2010,12 @@ public class TabletServer extends Abstra
// all for the same table id, so only need to read
// compaction id once
if (compactionId == null)
- compactionId = tablet.getCompactionID();
+ try {
+ compactionId = tablet.getCompactionID();
+ } catch (NoNodeException e) {
+ log.info("Asked to compact table with no compaction id " + ke + " " + e.getMessage());
+ return;
+ }
tablet.compactAll(compactionId);
}
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Module.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Module.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Module.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Module.java Tue Feb 7 20:59:36 2012
@@ -194,12 +194,12 @@ public class Module extends Node {
boolean test = false;
if (initNode instanceof Test) {
- startTimer(initNode.toString());
+ startTimer(initNode);
test = true;
}
initNode.visit(state, getProps(initNodeId));
if (test)
- stopTimer(initNode.toString());
+ stopTimer(initNode);
state.visitedNode();
// update aliases
@@ -245,12 +245,12 @@ public class Module extends Node {
try {
test = false;
if (nextNode instanceof Test) {
- startTimer(nextNode.toString());
+ startTimer(nextNode);
test = true;
}
nextNode.visit(state, nodeProps);
if (test)
- stopTimer(nextNode.toString());
+ stopTimer(nextNode);
} catch (Exception e) {
log.debug("Properties for node: " + nextNodeId);
for (Entry<Object,Object> entry : nodeProps.entrySet()) {
@@ -283,32 +283,37 @@ public class Module extends Node {
/**
*
*/
- private void startTimer(final String nodeName) {
+ private void startTimer(final Node initNode) {
runningLong.set(false);
timer = new Thread(new Runnable() {
@Override
public void run() {
- try {
- systemTime = System.currentTimeMillis();
- synchronized (timer) {
- timer.wait(time);
+ while (!runningLong.get()) {
+ try {
+ systemTime = System.currentTimeMillis();
+ synchronized (timer) {
+ timer.wait(time);
+ }
+ } catch (InterruptedException ie) {
+ return;
}
- } catch (InterruptedException ie) {
- return;
}
- log.warn("Node " + nodeName + " has been running for " + time / 1000.0 + " seconds. You may want to look into it.");
+ long timeSinceLastProgress = System.currentTimeMillis() - initNode.lastProgress();
+ if (timeSinceLastProgress > time) {
+ log.warn("Node " + initNode + " has been running for " + timeSinceLastProgress / 1000.0 + " seconds. You may want to look into it.");
+ }
runningLong.set(true);
}
-
});
+ initNode.makingProgress();
timer.start();
}
/**
*
*/
- private void stopTimer(String nodeName) {
+ private void stopTimer(Node nextNode) {
synchronized (timer) {
timer.interrupt();
try {
@@ -319,7 +324,7 @@ public class Module extends Node {
}
}
if (runningLong.get())
- log.warn("Node " + nodeName + ", which was running long, has now completed after " + (System.currentTimeMillis() - systemTime) / 1000.0 + " seconds");
+ log.warn("Node " + nextNode + ", which was running long, has now completed after " + (System.currentTimeMillis() - systemTime) / 1000.0 + " seconds");
}
@Override
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Node.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Node.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Node.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/Node.java Tue Feb 7 20:59:36 2012
@@ -26,6 +26,7 @@ import org.apache.log4j.Logger;
public abstract class Node {
protected final Logger log = Logger.getLogger(this.getClass());
+ long progress = System.currentTimeMillis();
/**
* Visits node
@@ -45,4 +46,12 @@ public abstract class Node {
public int hashCode() {
return toString().hashCode();
}
+
+ synchronized public void makingProgress() {
+ progress = System.currentTimeMillis();
+ }
+
+ synchronized public long lastProgress() {
+ return progress;
+ }
}
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/bulk/Verify.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/bulk/Verify.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/bulk/Verify.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/bulk/Verify.java Tue Feb 7 20:59:36 2012
@@ -45,8 +45,13 @@ public class Verify extends Test {
public void visit(State state, Properties props) throws Exception {
ThreadPoolExecutor threadPool = Setup.getThreadPool(state);
threadPool.shutdown();
+ int lastSize = 0;
while (!threadPool.isTerminated()) {
- log.info("Waiting for " + (threadPool.getQueue().size() + threadPool.getActiveCount()) + " nodes to complete");
+ int size = threadPool.getQueue().size() + threadPool.getActiveCount();
+ log.info("Waiting for " + size + " nodes to complete");
+ if (size != lastSize)
+ makingProgress();
+ lastSize = size;
threadPool.awaitTermination(10, TimeUnit.SECONDS);
}
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/BatchScan.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/BatchScan.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/BatchScan.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/BatchScan.java Tue Feb 7 20:59:36 2012
@@ -24,6 +24,7 @@ import java.util.Properties;
import java.util.Random;
import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.TableDeletedException;
@@ -71,6 +72,12 @@ public class BatchScan extends Test {
log.debug("BatchScan " + tableName + " failed, table deleted");
} catch (TableOfflineException e) {
log.debug("BatchScan " + tableName + " failed, offline");
+ } catch (RuntimeException e) {
+ if (e.getCause() instanceof AccumuloSecurityException) {
+ log.debug("BatchScan " + tableName + " failed, permission error");
+ } else {
+ throw e;
+ }
}
}
}
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/ScanTable.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/ScanTable.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/ScanTable.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/concurrent/ScanTable.java Tue Feb 7 20:59:36 2012
@@ -23,6 +23,7 @@ import java.util.Properties;
import java.util.Random;
import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.TableDeletedException;
@@ -59,6 +60,12 @@ public class ScanTable extends Test {
log.debug("Scan " + tableName + " failed, doesnt exist");
} catch (TableOfflineException e) {
log.debug("Scan " + tableName + " failed, offline");
+ } catch (RuntimeException e) {
+ if (e.getCause() instanceof AccumuloSecurityException) {
+ log.debug("BatchScan " + tableName + " failed, permission error");
+ } else {
+ throw e;
+ }
}
}
}
Modified: incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/shard/VerifyIndex.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/shard/VerifyIndex.java?rev=1241622&r1=1241621&r2=1241622&view=diff
==============================================================================
--- incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/shard/VerifyIndex.java (original)
+++ incubator/accumulo/trunk/src/server/src/main/java/org/apache/accumulo/server/test/randomwalk/shard/VerifyIndex.java Tue Feb 7 20:59:36 2012
@@ -17,8 +17,8 @@
package org.apache.accumulo.server.test.randomwalk.shard;
import java.util.Iterator;
-import java.util.Properties;
import java.util.Map.Entry;
+import java.util.Properties;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.Scanner;
@@ -54,6 +54,8 @@ public class VerifyIndex extends Test {
if (key1.compareTo(key2, PartialKey.ROW_COLFAM_COLQUAL) != 0)
throw new Exception("index rebuild mismatch " + key1 + " " + key2 + " " + indexTableName + " " + tmpIndexTableName);
count++;
+ if (count % 1000 == 0)
+ makingProgress();
}
if (iter.hasNext())